fix wrong version introduced by doxtop

.github/workflows/elixir.yml

@@ -2,12 +2,12 @@ name: mix
 on: push
 jobs:
   build:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     steps:
     - uses: actions/checkout@v2
     - uses: erlef/setup-elixir@v1
       with:
-        otp-version: 23.x
+        otp-version: 22.x
         elixir-version: 1.9.x
     - name: Dependencies
       run: |

.gitignore

@@ -7,3 +7,4 @@ rocksdb/
 deps/
 _build/
 *.lock
+doc/

LICENSE

@@ -16,7 +16,7 @@ OR IN INVIDIVUAL MANNER.
 
 YOU CANNOT USE THIS SOFTWARE BY ANY MEANS IN INTEREST OF LEGAL
 ENTITIES OR INDIVIDUALS WHO IS SUPPORTING NOW OR WAS SUPPORTING
-BACK THEN FASCISM, RUSCISM, COMMUNISM, CHOVINISM, HUMILIATION,
+BACK THEN FASCISM, RUSCISM, COMMUNISM, CHAUVINISM, HUMILIATION,
 AND OTHER SUPPRESSIVE IDEOLOGIES IN DIFFERENT EXPRESSIONS.
 
 STOP KILLING UKRAINIANS,

include/api.hrl

@@ -1,7 +1,9 @@
 -ifndef(API_HRL).
 -define(API_HRL, true).
 -define(API,[start/0,stop/0,leave/0,leave/1,destroy/0,destroy/1,
-             join/0,join/1,modules/0,cursors/0,get/2,get/3,put/1,put/2,index/3,index/4,match/1,match/2,index_match/2,index_match/3,delete/2,delete/3,
+             join/0,join/1,modules/0,cursors/0,get/2,get/3,put/1,put/2,index/3,index/4,
+             match/1,match/2,index_match/2,index_match/3,key_match/2,key_match/3,
+             delete/2,delete/3,delete_range/3,
              table/1,tables/0,dir/0,initialize/2,seq/2,all/1,all/2,count/1,ver/0]).
 -include("metainfo.hrl").
 -spec seq(atom() | [], integer() | []) -> term().

include/backend.hrl

@@ -1,13 +1,15 @@
 -ifndef(BACKEND_HRL).
 -define(BACKEND_HRL, true).
--define(BACKEND, [db/0,get/3,put/1,put/2,delete/3,index/3,dump/0,start/0,stop/0,destroy/0,destroy/1,
-                  join/2,leave/0,leave/1,dir/0,create_table/2,add_table_index/2,seq/2,all/2,count/1,version/0,match/1,index_match/2]).
+-define(BACKEND, [db/0,get/3,put/1,put/2,delete/3,delete_range/3,index/3,dump/0,start/0,stop/0,destroy/0,destroy/1,keys/2,
+                  join/2,leave/0,leave/1,dir/0,create_table/2,add_table_index/2,seq/2,all/2,count/1,version/0,
+                  match/1,key_match/3,index_match/2]).
 -compile({no_auto_import,[get/1,put/2]}).
 -include("kvs.hrl").
 -spec put(tuple() | list(tuple())) -> ok | {error,any()}.
 -spec put(tuple() | list(tuple()), #kvs{}) -> ok | {error,any()}.
 -spec get(term() | any(), any(), #kvs{}) -> {ok,any()} | {error,not_found}.
 -spec delete(term(), any(), #kvs{}) -> ok | {error,not_found}.
+-spec delete_range(term(), any(), #kvs{}) -> ok | {error,not_found}.
 -spec dump() -> ok.
 -spec start() -> ok.
 -spec stop() -> ok.

include/stream.hrl

@@ -3,7 +3,7 @@
 -include("kvs.hrl").
 -include("cursors.hrl").
 -define(STREAM, [top/1, top/2, bot/1, bot/2, next/1, next/2, prev/1, prev/2, drop/1, drop/2, take/1, take/2, append/2, append/3, feed/1, feed/2,
-                 load_reader/1, load_reader/2, writer/1, writer/2, reader/1, reader/2, save/1, save/2, add/1, add/2, remove/2, remove/3]).
+                 load_reader/1, load_reader/2, writer/1, writer/2, reader/1, reader/2, save/1, save/2, add/1, add/2, remove/2, remove/3, cut/1, cut/2]).
 
 -spec top(#reader{})  -> #reader{}.
 -spec bot(#reader{})  -> #reader{}.
@@ -19,4 +19,5 @@
 -spec add(#writer{}) -> #writer{}.
 -spec append(tuple(),term()) -> any().
 -spec remove(tuple(),term()) -> integer().
+-spec cut(term(), list()) -> ok.
 -endif.

src/kvs.erl

@@ -24,11 +24,12 @@
          ensure/1,
          ensure/2,
          seq_gen/0,
+         keys/1,
          fields/1,
          defined/2,
          field/2,
          setfield/3,
-         cut/2]).
+         remove/1]).
 
 -export([join/2, seq/3]).
 
@@ -73,6 +74,12 @@ get(Table, Key) ->
 index(Table, K, V) ->
     index(Table, K, V, #kvs{mod = dba()}).
 
+keys(Feed) ->
+    keys(Feed, #kvs{mod = dba(), db = db()}).
+
+key_match(Feed, Id) ->
+  key_match(Feed, Id, #kvs{mod = dba(), db=db()}).
+
 match(Record) ->
     match(Record, #kvs{mod = dba()}).
 
@@ -120,11 +127,20 @@ get(RecordName, Key, #kvs{mod = Mod, db = Db}) ->
 delete(Tab, Key, #kvs{mod = Mod, db = Db}) ->
     Mod:delete(Tab, Key, Db).
 
+delete_range(Feed, Last, #kvs{mod=DBA, db=Db}) ->
+    DBA:delete_range(Feed,Last,Db).
+
 count(Tab, #kvs{mod = DBA}) -> DBA:count(Tab).
 
 index(Tab, Key, Value, #kvs{mod = DBA}) ->
     DBA:index(Tab, Key, Value).
 
+keys(Feed, #kvs{mod = DBA, db = Db}) ->
+    DBA:keys(Feed, Db).
+
+key_match(Feed, Id, #kvs{mod = DBA, db = Db}) ->
+    DBA:key_match(Feed, Id, Db).
+
 match(Record, #kvs{mod = DBA}) ->
     DBA:match(Record).
 
@@ -186,7 +202,11 @@ save(X)                      -> (kvs_stream()):save(X).
 
 save(X,#kvs{db = Db})        -> (kvs_stream()):save(X,Db).
 
-cut(X, Y)                    -> (kvs_stream()):cut(X, Y).
+remove(X)                    -> (kvs_stream()):remove(X).
+
+cut(X)                       -> (kvs_stream()):cut(X).
+
+cut(X,#kvs{db = Db})         -> (kvs_stream()):cut(X, Db).
 
 add(X)                       -> (kvs_stream()):add(X).
 

src/stores/kvs_st.erl → src/layers/kvs_st.erl

@@ -3,8 +3,8 @@
 -include("stream.hrl").
 -include("metainfo.hrl").
 -export(?STREAM).
--import(kvs_rocks, [key/2, key/1, bt/1, tb/1, ref/0, ref/1, seek_it/1, seek_it/2, move_it/3, move_it/4, take_it/4, take_it/5, estimate/0, estimate/1]).
--export([raw_append/2,raw_append/3]).
+-import(kvs_rocks, [fmt/1, key/2, key/1, bt/1, tb/1, ref/0, ref/1, seek_it/1, seek_it/2, move_it/3, move_it/4, take_it/4, take_it/5, delete_it/2, estimate/0, estimate/1]).
+-export([raw_append/2,raw_append/3, remove/1]).
 
 db() -> application:get_env(kvs,rocks_name,"rocksdb").
 
@@ -41,6 +41,9 @@ take(#reader{args=N,feed=Feed,cache=I,dir=_}=C,Db) -> read_it(C,take_it(k(Feed,I
 drop(#reader{}=X) -> drop(X,db()).
 drop(#reader{args=N}=C,_) when N =< 0 -> C;
 drop(#reader{}=C,Db) -> (take(C#reader{dir=0},Db))#reader{args=[]}.
+remove(#reader{}=C) -> remove(C, db()).
+remove(#reader{feed=Feed}=C,Db) -> R = read_it(C, delete_it(Feed,Db)), kvs:delete(writer, Feed), R;
+remove(Rec,Feed) -> remove(Rec,Feed,db()).
 
 feed(Feed) -> feed(Feed,db()).
 feed(Feed,Db) ->
@@ -93,7 +96,14 @@ add(M,#writer{id=Feed,count=S}=C,Db) ->
    raw_append(M,Feed,Db),
    C#writer{cache={e(1,M),e(2,M),key(Feed)},count=NS}.
 
-remove(Rec,Feed) -> remove(Rec,Feed,db()).
+cut(Feed) -> cut(Feed,db()).
+cut(Feed,Db) ->
+  #writer{cache={_,Key,Fd}=Ch} = kvs:writer(Feed, #kvs{db=Db,mod=kvs_rocks}),
+  #reader{} = kvs:prev(reader(Feed, Db)),
+  #reader{} = kvs:next(#reader{feed=key(Feed), cache=Ch}),
+
+  kvs:delete_range(Feed,{Fd,Key},#kvs{db=Db,mod=kvs_rocks}).
+
 remove(Rec,Feed,Db) ->
   kvs:ensure(#writer{id=Feed},#kvs{db=Db,mod=kvs_rocks}),
   W = #writer{count=C, cache=Ch} = kvs:writer(Feed,#kvs{db=Db,mod=kvs_rocks}),

src/layers/kvs_stream.erl

@@ -107,6 +107,8 @@ add(M,#writer{cache=V1,count=S}=C,Db) ->
     N=sp(sn(M,[]),id(V)), P=sn(V,id(M)), kvs:put([N,P],#kvs{db=Db}),
     C#writer{cache=N,count=S+1}.
 
+cut(Feed) -> cut(Feed, db()).
+cut(_,_) -> ignore.
 remove(Rec,Feed)  -> remove(Rec,Feed,db()).
 remove(_Rec,Feed,Db) ->
    {ok,W=#writer{count=Count}} = kvs:get(writer,Feed,#kvs{db=Db}),

src/stores/kvs_fs.erl

@@ -75,6 +75,9 @@ delete(TableName, Key, _) ->
             file:delete(File);
         {error,X} -> {error,X}
     end.
+delete_range(_,_,_) -> {error, not_found}.
+keys(_,_) -> [].
+key_match(_,_,_) -> [].
 
 count(RecordName) -> length(filelib:fold_files(filename:join([dir_name(), RecordName]), "",true, fun(A,Acc)-> [A|Acc] end, [])).
 

src/stores/kvs_mnesia.erl

@@ -36,6 +36,10 @@ initialize() ->
 index(Tab,Key,Value) ->
     lists:flatten(many(fun() -> mnesia:index_read(Tab,Value,Key) end)).
 
+keys(Tab,_) -> mnesia:all_keys(Tab).
+
+key_match(_Tab,_Id,_) -> [].
+
 get(RecordName, Key, _) -> just_one(fun() -> mnesia:read(RecordName, Key) end).
 put(R)                  -> put(R,db()).
 put(Records, _) when is_list(Records) -> void(fun() -> lists:foreach(fun mnesia:write/1, Records) end);
@@ -45,6 +49,7 @@ delete(Tab, Key, _) ->
         {aborted,Reason} -> {error,Reason};
         {atomic,_Result} -> ok;
         _ -> ok end.
+delete_range(_,_,_) -> {error, not_found}.
 match(Record) -> lists:flatten(many(fun() -> mnesia:match_object(Record) end)).
 index_match(Record, Index) -> lists:flatten(many(fun() -> mnesia:index_match_object(Record, Index) end)).
 count(RecordName) -> mnesia:table_info(RecordName, size).

src/stores/kvs_rocks.erl

@@ -5,7 +5,7 @@
 -include_lib("stdlib/include/qlc.hrl").
 -export(?BACKEND).
 -export([ref/0,ref/1,bt/1,key/2,key/1,fd/1,tb/1,estimate/0,estimate/1]).
--export([seek_it/1, seek_it/2, move_it/3, move_it/4, take_it/4, take_it/5]).
+-export([seek_it/1, seek_it/2, move_it/3, move_it/4, take_it/4, take_it/5, delete_it/1, delete_it/2]).
 
 e(X,Y) -> element(X,Y).
 
@@ -26,6 +26,35 @@ key(writer,R) -> % allow old writers
               iolist_to_binary([lists:join(<<"/">>, lists:flatten([<<>>, erlang:atom_to_binary(writer, utf8), tb(R)]))]);
 key(Tab,R) -> Fd = case Tab of [] -> []; _ -> tb(Tab) end,
               iolist_to_binary([lists:join(<<"/">>, lists:flatten([<<>>, Fd, fmt(R)]))]).
+keys(Tab, Db) ->
+    Feed = key(Tab,[]),
+    {ok, H} = rocksdb:iterator(ref(Db), []),
+    Keys = fun KEY(K1,Acc) when binary_part(K1,{0,byte_size(Feed)}) =:= Feed ->
+                  case rocksdb:iterator_move(H, next) of
+                    {ok,K2,_} -> KEY(K2,[tb(K1)|Acc]);
+                            _ -> lists:reverse([tb(K1)|Acc])
+                  end;
+               KEY(_,Acc) -> rocksdb:iterator_close(H), lists:reverse(Acc)
+           end,
+    {ok, K, _} = rocksdb:iterator_move(H, {seek, Feed}),
+    Keys(K,[]).
+
+ key_match(Tab, Id, Db) ->
+  Feed = key(Tab,[]),
+  {ok, H} = rocksdb:iterator(ref(Db), []),
+  Keys = fun KEY(K1) when 
+              binary_part(K1,{0,byte_size(Feed)}) =:= Feed andalso
+              binary_part(K1,{byte_size(K1), -byte_size(Id)}) =:= Id ->
+              rocksdb:iterator_close(H), [K1];
+             KEY(K1) when binary_part(K1,{0,byte_size(Feed)}) =:= Feed ->
+                case rocksdb:iterator_move(H, next) of
+                  {ok,K2,_} -> KEY(K2);
+                          _ -> []
+                end;
+             KEY(_) -> rocksdb:iterator_close(H), []
+         end,
+  {ok, K, _} = rocksdb:iterator_move(H, {seek, Feed}),
+  Keys(K).
 
 fmt([]) -> [];
 fmt(K) -> Key = tb(K),
@@ -111,10 +140,27 @@ join(_,Db)       ->
               initialize(),
               application:set_env(kvs,ref_env(Db),Ref).
 
-compile(seek) -> [fun rocksdb:iterator/2,fun rocksdb:iterator_move/2];
-compile(move) -> [fun rocksdb:iterator_move/2];
-compile(close) -> [fun rocksdb:iterator_close/1].
+compile(it)     -> [fun rocksdb:iterator/2];
+compile(seek)   -> [fun rocksdb:iterator/2,fun rocksdb:iterator_move/2];
+compile(move)   -> [fun rocksdb:iterator_move/2];
+compile(close)  -> [fun rocksdb:iterator_close/1].
 compile(take,N) -> lists:map(fun(_) -> fun rocksdb:iterator_move/2 end, lists:seq(1, N)).
+compile(delete,_, {error,E},_) -> {error,E};
+compile(delete,SK,{ok,_,V1,_},Db) ->
+  F1 = key(key(fmt(SK),e(2,V1))), S = sz(SK),
+  [fun Del(H,Dir) ->
+    case rocksdb:delete(ref(Db), F1, []) of ok ->      
+      % {ok, K} case exist only in api, but never actually used
+      case rocksdb:iterator_move(H,Dir) of
+        {ok,K,_} when binary_part(K,{0,S}) == SK -> case rocksdb:delete(ref(Db), K, []) of ok -> Del(H,Dir); E -> E end;
+        {ok,K}   when binary_part(K,{0,S}) == SK -> case rocksdb:delete(ref(Db), K, []) of ok -> Del(H,Dir); E -> E end;
+        {ok,K,V} -> {ok,K,V};
+        {ok,K}   -> {ok, K};
+        E -> E
+      end;
+      E -> E
+    end
+  end].
 
 stop_it(H) -> try begin [F]=compile(close), F(H) end catch error:badarg -> ok end.
 seek_it(K) -> seek_it(K,db()).
@@ -124,6 +170,8 @@ move_it(Key,SK,Dir,Db) -> run(Key,SK,Dir,compile(seek) ++ compile(move),Db).
 take_it(Key,SK,Dir,N) -> take_it(Key,SK,Dir,N,db()).
 take_it(Key,SK,Dir,N,Db) when is_integer(N) andalso N >= 0 -> run(Key,SK,Dir,compile(seek) ++ compile(take,N),Db);
 take_it(Key,SK,Dir,_,Db) -> take_it(Key,SK,Dir,0,Db).
+delete_it(Fd) -> delete_it(Fd, db()).
+delete_it(Fd,Db) -> run(Fd,Fd,next,compile(seek) ++ compile(delete,Fd,seek_it(Fd),Db),Db).
 
 all(R,Db) -> kvs_st:feed(R,Db).
 
@@ -138,6 +186,25 @@ put(Record) -> put(Record,db()).
 put(Records,Db) when is_list(Records) -> lists:map(fun(Record) -> put(Record,Db) end, Records);
 put(Record,Db) -> rocksdb:put(ref(Db), key(Record), term_to_binary(Record), [{sync,true}]).
 delete(Feed, Id, Db) -> rocksdb:delete(ref(Db), key(Feed,Id), []).
+delete_range(Feed,{Fd,Key},Db) ->
+  Last = key(key(fmt(Fd),Key)),
+  ReadOps = [{'prefix_same_as_start', true}],
+  CompactOps = [{change_level, true}],
+  Feed1 = key(Feed),
+  Sz = size(Feed1),
+  Reopen = case ref(Db) of [] -> skip; _ -> leave(Db), ok end,
+
+  {ok, R} = rocksdb:open(Db, [{prefix_extractor, {capped_prefix_transform, Sz}}]),
+  {ok, H} = rocksdb:iterator(R, ReadOps),
+  {ok, Start, _} = rocksdb:iterator_move(H, {seek, Feed1}),
+
+  ok = rocksdb:delete_range(R, Start, Last, []),
+  ok = rocksdb:delete(R, Last, []),
+  ok = rocksdb:delete(R, key(writer,Feed), []),
+  ok = rocksdb:compact_range(R, Start, undefined, CompactOps),
+  ok = rocksdb:iterator_close(H),
+  ok = rocksdb:close(R),
+  case Reopen of skip -> ok; ok -> join([],Db) end.
 count(_) -> 0.
 estimate()   -> estimate(db()).
 estimate(Db) -> case rocksdb:get_property(ref(Db), <<"rocksdb.estimate-num-keys">>) of

test/fd_test.exs

@@ -17,7 +17,7 @@ defmodule Fd.Test do
 
     defrecord(:msg, id: [], body: [])
 
-    setup do: (on_exit(fn -> :ok = :kvs.leave();:ok = :kvs.destroy() end);:kvs.join())
+    setup do: (on_exit(fn -> :kvs.leave();:ok = :kvs.destroy() end);:kvs.join())
     setup kvs, do: [
         id0: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/duck") end, :lists.seq(1,10)),
         id1: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/luck") end, :lists.seq(1,10)),
@@ -91,10 +91,10 @@ defmodule Fd.Test do
             end)
         r = :kvs.load_reader(rid)
         assert r = :kvs.prev(r)
-        assert r = KVS.reader(:kvs.prev(:kvs.top(r)), args: [])                 
+        assert r = KVS.reader(:kvs.prev(:kvs.top(r)), args: [])
     end
 
-    test "prev to empty" do        
+    test "prev to empty" do
         :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/aco") end, :lists.seq(1,2))
         all = :kvs.all("/aco")
         head = Enum.at(all,0)
@@ -105,6 +105,43 @@ defmodule Fd.Test do
         assert KVS.reader(args: [^head]) = :kvs.take(KVS.reader(r1, args: 1000, dir: 1))
     end
 
+    test "cut the *uck", kvs do
+        :kvs.cut("/crm/luck")
+
+        all = :kvs.all("/crm")
+        assert 20 = length(all)
+        assert all = :kvs.all("/crm/duck") ++ :kvs.all("/crm/truck")
+
+        :kvs.cut("/crm/duck")
+
+        all = :kvs.all("/crm")
+        assert 10 = length(all)
+        assert all = :kvs.all("/crm/truck")
+
+        :kvs.cut("/crm/truck")
+
+        all = :kvs.all("/crm")
+        assert 0 = length(all)
+    end
+
+    test "remove the *uck with readers", kvs do
+        :kvs.remove(:kvs.reader("/crm/luck"))
+
+        all = :kvs.all("/crm")
+        assert 20 = length(all)
+        assert all = :kvs.all("/crm/duck") ++ :kvs.all("/crm/truck")
+
+        :kvs.remove(:kvs.reader("/crm/duck"))
+
+        all = :kvs.all("/crm")
+        assert 10 = length(all)
+        assert all = :kvs.all("/crm/truck")
+
+        :kvs.remove(:kvs.reader("/crm/truck"))
+
+        all = :kvs.all("/crm")
+        assert 0 = length(all)
+    end
 
     @tag :skip # can`t manage this within current implementation. create correct keys!
     test "keys with feeds separator" do
@@ -115,17 +152,17 @@ defmodule Fd.Test do
 
     test "corrupted writers doesn't affect all" do
         prev = :kvs.all("/crm/duck")
-        
+
         KVS.writer(cache: ch) = w = :kvs.writer("/crm/duck")
         w1 = KVS.writer(w, cache: {:msg, "unknown", "/corrupted"})
-        
+
         :ok = :kvs_rocks.put(w1)
         w2 = :kvs.writer("/crm/duck")
         assert {:ok, ^w2} = :kvs.get(:writer, "/crm/duck")
         assert w1 == w2
 
         assert prev = :kvs.all("/crm/duck")
-        
+
         {:ok,_} = :kvs.get(:writer, "/crm/duck")
         :ok = :kvs.delete(:writer, "/crm/duck")
         {:error, :not_found} = :kvs.get(:writer, "/crm/duck")

test/st_test.exs

@@ -38,7 +38,7 @@ defmodule St.Test do
 
         assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.take(KVS.reader(r, args: 0, dir: 0))
         assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.take(KVS.reader(r, args: -1, dir: 0))
-        
+
         assert r1 = KVS.reader(id: ^rid, feed: ^feed, args: a01) = :kvs.take(KVS.reader(r,  args: 10, dir: 0))
         assert kvs[:id2] |> Enum.map(&msg(id: &1)) == a01
         assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.take(KVS.reader(r1, args: 10, dir: 0))
@@ -68,7 +68,7 @@ defmodule St.Test do
         assert KVS.reader(id: ^rid, feed: ^feed, args: ^tpm, dir: 1) = :kvs.take(KVS.reader(r, args: 100, dir: 1))
 
         assert r1 = KVS.reader(feed: ^feed, count: 10, args: [], cache: {:msg,^bot,^feed}) = :kvs.bot(r)
-        
+
         assert r2 = KVS.reader(feed: ^feed, count: 10, args: a01) = :kvs.take(KVS.reader(r1, args: 5, dir: 1))
         x01 = Enum.drop(kvs[:id1],5) |> Enum.map(&msg(id: &1)) |> Enum.reverse
         assert x01 == a01
@@ -114,7 +114,7 @@ defmodule St.Test do
         assert :kvs.feed("/crm/personal/Реєстратор А1/in/directory/duck") 
             ++ :kvs.feed("/crm/personal/Реєстратор А1/in/doc")
             ++ :kvs.feed("/crm/personal/Реєстратор А1/in/mail") == :kvs.feed("/crm/personal/Реєстратор А1/in")
-    end 
+    end
 
     defp log(x), do: IO.puts '#{inspect(x)}'
     defp log(m, x), do: IO.puts '#{m} #{inspect(x)}'