<commit_before>%%====================================================================
%%
%% LeoFS Storage
%%
%% Copyright (c) 2012 Rakuten, Inc.
%%
%% 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.
%%
%% -------------------------------------------------------------------
%% LeoFS Storage - MQ Client
%% @doc
%% @end
%%====================================================================
-module(leo_storage_mq_client).

-author('<NAME>').

-behaviour(leo_mq_behaviour).

-include("leo_storage.hrl").
-include_lib("leo_commons/include/leo_commons.hrl").
-include_lib("leo_logger/include/leo_logger.hrl").
-include_lib("leo_mq/include/leo_mq.hrl").
-include_lib("leo_object_storage/include/leo_object_storage.hrl").
-include_lib("leo_redundant_manager/include/leo_redundant_manager.hrl").
-include_lib("eunit/include/eunit.hrl").

-export([start/1, start/2, publish/3, publish/4, publish/5]).
-export([init/0, handle_call/1]).

-define(SLASH, "/").
-define(MSG_PATH_PER_OBJECT,     "1").
-define(MSG_PATH_SYNC_VNODE_ID,  "2").
-define(MSG_PATH_REBALANCE,      "3").
-define(MSG_PATH_ASYNC_DELETION, "4").

-type(queue_type() :: ?QUEUE_TYPE_PER_OBJECT  |
                      ?QUEUE_TYPE_SYNC_BY_VNODE_ID  |
                      ?QUEUE_TYPE_ASYNC_DELETION).

-type(queue_id()   :: ?QUEUE_ID_PER_OBJECT |
                      ?QUEUE_ID_SYNC_BY_VNODE_ID |
                      ?QUEUE_ID_REBALANCE |
                      ?QUEUE_ID_ASYNC_DELETION).

%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------
%% @doc create queues and launch mq-servers.
%%
start(RootPath0) ->
    start(leo_storage_sup, RootPath0).

-spec(start(pid(), string()) ->
             ok | {error, any()}).
start(RefSup, RootPath0) ->
    %% launch mq-sup under storage-sup
    RefMqSup =
        case whereis(leo_mq_sup) of
            undefined ->
                ChildSpec = {leo_mq_sup,
                             {leo_mq_sup, start_link, []},
                             permanent, 2000, supervisor, [leo_mq_sup]},
                {ok, Pid} = supervisor:start_child(RefSup, ChildSpec),
                Pid;
            Pid ->
                Pid
        end,

    %% launch queue-processes
    RootPath1 =
        case (string:len(RootPath0) == string:rstr(RootPath0, ?SLASH)) of
            true  -> RootPath0;
            false -> RootPath0 ++ ?SLASH
        end,

    ?TBL_REBALANCE_COUNTER = ets:new(?TBL_REBALANCE_COUNTER,
                                     [named_table, public, {read_concurrency, true}]),
    lists:foreach(
      fun({Id, Path, MaxInterval, MinInterval}) ->
              leo_mq_api:new(RefMqSup, Id, [{?MQ_PROP_MOD,          ?MODULE},
                                            {?MQ_PROP_FUN,          ?MQ_SUBSCRIBE_FUN},
                                            {?MQ_PROP_ROOT_PATH,    RootPath1 ++ Path},
                                            {?MQ_PROP_DB_PROCS,     ?env_num_of_mq_procs()},
                                            {?MQ_PROP_MAX_INTERVAL, MaxInterval},
                                            {?MQ_PROP_MIN_INTERVAL, MinInterval}
                                           ])
      end, [{?QUEUE_ID_PER_OBJECT,        ?MSG_PATH_PER_OBJECT,      64, 16},
            {?QUEUE_ID_SYNC_BY_VNODE_ID,  ?MSG_PATH_SYNC_VNODE_ID,   32,  8},
            {?QUEUE_ID_REBALANCE,         ?MSG_PATH_REBALANCE,       32,  8},
            {?QUEUE_ID_ASYNC_DELETION,    ?MSG_PATH_ASYNC_DELETION, 128, 64}
           ]),
    ok.

%% @doc Input a message into the queue.
%%
-spec(publish(queue_type(), integer(), atom()) ->
             ok).
publish(?QUEUE_TYPE_SYNC_BY_VNODE_ID = Id, VNodeId, Node) ->
    KeyBin     = term_to_binary(VNodeId),
    MessageBin = term_to_binary(#sync_unit_of_vnode_message{id        = leo_date:clock(),
                                                            vnode_id  = VNodeId,
                                                            node      = Node,
                                                            timestamp = leo_date:now()}),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(?QUEUE_TYPE_ASYNC_DELETION = Id, AddrId, Key) ->
    KeyBin     = term_to_binary({AddrId, Key}),
    MessageBin = term_to_binary(#async_deletion_message{id        = leo_date:clock(),
                                                        addr_id   = AddrId,
                                                        key       = Key,
                                                        timestamp = leo_date:now()}),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(_,_,_) ->
    {error, badarg}.

-spec(publish(queue_type(), integer(), any(), any()) ->
             ok).
publish(?QUEUE_TYPE_PER_OBJECT = Id, AddrId, Key, ErrorType) ->
    KeyBin = term_to_binary({ErrorType, Key}),
    PublishedAt = leo_date:now(),
    MessageBin  = term_to_binary(#inconsistent_data_message{id        = leo_date:clock(),
                                                            type      = ErrorType,
                                                            addr_id   = AddrId,
                                                            key       = Key,
                                                            timestamp = PublishedAt}),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(_,_,_,_) ->
    {error, badarg}.

publish(?QUEUE_TYPE_REBALANCE = Id, Node, VNodeId, AddrId, Key) ->
    KeyBin     = term_to_binary({Node, AddrId, Key}),
    MessageBin = term_to_binary(#rebalance_message{id        = leo_date:clock(),
                                                   vnode_id  = VNodeId,
                                                   addr_id   = AddrId,
                                                   key       = Key,
                                                   node      = Node,
                                                   timestamp = leo_date:now()}),
    Table = ?TBL_REBALANCE_COUNTER,
    case ets_lookup(Table, VNodeId) of
        {ok, 0} -> ets:insert(Table, {VNodeId, 0});
        _Other  -> void
    end,
    ok = increment_counter(Table, VNodeId),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(_,_,_,_,_) ->
    {error, badarg}.


%% -------------------------------------------------------------------
%% Callbacks
%% -------------------------------------------------------------------
%% @doc Initializer
%%
-spec(init() ->
             ok | {error, any()}).
init() ->
    ok.


%% @doc Subscribe a message from the queue.
%%
-spec(handle_call({publish | consume, any() | queue_id() , any() | binary()}) ->
             ok | {error, any()}).
handle_call({publish, _Id, _Reply}) ->
    ok;

handle_call({consume, ?QUEUE_ID_PER_OBJECT, MessageBin}) ->
    case catch binary_to_term(MessageBin) of
        {'EXIT', Cause} ->
            {error, Cause};
        #inconsistent_data_message{addr_id = AddrId,
                                   key     = Key,
                                   type    = ErrorType} ->
            case correct_redundancies(Key) of
                ok ->
                    ok;
                Error ->
                    publish(?QUEUE_TYPE_PER_OBJECT, AddrId, Key, ErrorType),
                    Error
            end
    end;

handle_call({consume, ?QUEUE_ID_SYNC_BY_VNODE_ID, MessageBin}) ->
    case catch binary_to_term(MessageBin) of
        {'EXIT', Cause} ->
            {error, Cause};
        #sync_unit_of_vnode_message{vnode_id = ToVNodeId,
                                    node     = Node} ->
            case leo_redundant_manager_api:range_of_vnodes(ToVNodeId) of
                {ok, Res} ->
                    {ok, {CurRingHash, _PrevRingHash}} =
                        leo_redundant_manager_api:checksum(?CHECKSUM_RING),
                    ok = sync_vnodes(Node, CurRingHash, Res),
                    notify_rebalance_message_to_manager(ToVNodeId);
                Error ->
                    Error
            end
    end;

handle_call({consume, ?QUEUE_ID_REBALANCE, MessageBin}) ->
    case catch binary_to_term(MessageBin) of
        {'EXIT', Cause} ->
            {error, Cause};
        #rebalance_message{node     = Node,
                           vnode_id = VNodeId,
                           addr_id  = AddrId,<commit_msg>Fix mq-consumer's bug - Registered ignorance of messages into the queue<commit_after>%%====================================================================
%%
%% LeoFS Storage
%%
%% Copyright (c) 2012 Rakuten, Inc.
%%
%% 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.
%%
%% -------------------------------------------------------------------
%% LeoFS Storage - MQ Client
%% @doc
%% @end
%%====================================================================
-module(leo_storage_mq_client).

-author('<NAME>').

-behaviour(leo_mq_behaviour).

-include("leo_storage.hrl").
-include_lib("leo_commons/include/leo_commons.hrl").
-include_lib("leo_logger/include/leo_logger.hrl").
-include_lib("leo_mq/include/leo_mq.hrl").
-include_lib("leo_object_storage/include/leo_object_storage.hrl").
-include_lib("leo_redundant_manager/include/leo_redundant_manager.hrl").
-include_lib("eunit/include/eunit.hrl").

-export([start/1, start/2, publish/3, publish/4, publish/5]).
-export([init/0, handle_call/1]).

-define(SLASH, "/").
-define(MSG_PATH_PER_OBJECT,     "1").
-define(MSG_PATH_SYNC_VNODE_ID,  "2").
-define(MSG_PATH_REBALANCE,      "3").
-define(MSG_PATH_ASYNC_DELETION, "4").

-type(queue_type() :: ?QUEUE_TYPE_PER_OBJECT  |
                      ?QUEUE_TYPE_SYNC_BY_VNODE_ID  |
                      ?QUEUE_TYPE_ASYNC_DELETION).

-type(queue_id()   :: ?QUEUE_ID_PER_OBJECT |
                      ?QUEUE_ID_SYNC_BY_VNODE_ID |
                      ?QUEUE_ID_REBALANCE |
                      ?QUEUE_ID_ASYNC_DELETION).

%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------
%% @doc create queues and launch mq-servers.
%%
start(RootPath0) ->
    start(leo_storage_sup, RootPath0).

-spec(start(pid(), string()) ->
             ok | {error, any()}).
start(RefSup, RootPath0) ->
    %% launch mq-sup under storage-sup
    RefMqSup =
        case whereis(leo_mq_sup) of
            undefined ->
                ChildSpec = {leo_mq_sup,
                             {leo_mq_sup, start_link, []},
                             permanent, 2000, supervisor, [leo_mq_sup]},
                {ok, Pid} = supervisor:start_child(RefSup, ChildSpec),
                Pid;
            Pid ->
                Pid
        end,

    %% launch queue-processes
    RootPath1 =
        case (string:len(RootPath0) == string:rstr(RootPath0, ?SLASH)) of
            true  -> RootPath0;
            false -> RootPath0 ++ ?SLASH
        end,

    ?TBL_REBALANCE_COUNTER = ets:new(?TBL_REBALANCE_COUNTER,
                                     [named_table, public, {read_concurrency, true}]),
    lists:foreach(
      fun({Id, Path, MaxInterval, MinInterval}) ->
              leo_mq_api:new(RefMqSup, Id, [{?MQ_PROP_MOD,          ?MODULE},
                                            {?MQ_PROP_FUN,          ?MQ_SUBSCRIBE_FUN},
                                            {?MQ_PROP_ROOT_PATH,    RootPath1 ++ Path},
                                            {?MQ_PROP_DB_PROCS,     ?env_num_of_mq_procs()},
                                            {?MQ_PROP_MAX_INTERVAL, MaxInterval},
                                            {?MQ_PROP_MIN_INTERVAL, MinInterval}
                                           ])
      end, [{?QUEUE_ID_PER_OBJECT,        ?MSG_PATH_PER_OBJECT,      64, 16},
            {?QUEUE_ID_SYNC_BY_VNODE_ID,  ?MSG_PATH_SYNC_VNODE_ID,   32,  8},
            {?QUEUE_ID_REBALANCE,         ?MSG_PATH_REBALANCE,       32,  8},
            {?QUEUE_ID_ASYNC_DELETION,    ?MSG_PATH_ASYNC_DELETION, 128, 64}
           ]),
    ok.

%% @doc Input a message into the queue.
%%
-spec(publish(queue_type(), integer(), atom()) ->
             ok).
publish(?QUEUE_TYPE_SYNC_BY_VNODE_ID = Id, VNodeId, Node) ->
    KeyBin     = term_to_binary(VNodeId),
    MessageBin = term_to_binary(#sync_unit_of_vnode_message{id        = leo_date:clock(),
                                                            vnode_id  = VNodeId,
                                                            node      = Node,
                                                            timestamp = leo_date:now()}),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(?QUEUE_TYPE_ASYNC_DELETION = Id, AddrId, Key) ->
    KeyBin     = term_to_binary({AddrId, Key}),
    MessageBin = term_to_binary(#async_deletion_message{id        = leo_date:clock(),
                                                        addr_id   = AddrId,
                                                        key       = Key,
                                                        timestamp = leo_date:now()}),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(_,_,_) ->
    {error, badarg}.

-spec(publish(queue_type(), integer(), any(), any()) ->
             ok).
publish(?QUEUE_TYPE_PER_OBJECT = Id, AddrId, Key, ErrorType) ->
    KeyBin = term_to_binary({ErrorType, Key}),
    PublishedAt = leo_date:now(),
    MessageBin  = term_to_binary(#inconsistent_data_message{id        = leo_date:clock(),
                                                            type      = ErrorType,
                                                            addr_id   = AddrId,
                                                            key       = Key,
                                                            timestamp = PublishedAt}),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(_,_,_,_) ->
    {error, badarg}.

publish(?QUEUE_TYPE_REBALANCE = Id, Node, VNodeId, AddrId, Key) ->
    KeyBin     = term_to_binary({Node, AddrId, Key}),
    MessageBin = term_to_binary(#rebalance_message{id        = leo_date:clock(),
                                                   vnode_id  = VNodeId,
                                                   addr_id   = AddrId,
                                                   key       = Key,
                                                   node      = Node,
                                                   timestamp = leo_date:now()}),
    Table = ?TBL_REBALANCE_COUNTER,
    case ets_lookup(Table, VNodeId) of
        {ok, 0} -> ets:insert(Table, {VNodeId, 0});
        _Other  -> void
    end,
    ok = increment_counter(Table, VNodeId),
    leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);

publish(_,_,_,_,_) ->
    {error, badarg}.


%% -------------------------------------------------------------------
%% Callbacks
%% -------------------------------------------------------------------
%% @doc Initializer
%%
-spec(init() ->
             ok | {error, any()}).
init() ->
    ok.


%% @doc Subscribe a message from the queue.
%%
-spec(handle_call({publish | consume, any() | queue_id() , any() | binary()}) ->
             ok | {error, any()}).
handle_call({publish, _Id, _Reply}) ->
    ok;

handle_call({consume, ?QUEUE_ID_PER_OBJECT, MessageBin}) ->
    case catch binary_to_term(MessageBin) of
        {'EXIT', Cause} ->
            {error, Cause};
        #inconsistent_data_message{addr_id = AddrId,
                                   key     = Key,
                                   type    = ErrorType} ->
            case correct_redundancies(Key) of
                ok ->
                    ok;
                {error, not_found} ->
                    ok;
                Error ->
                    publish(?QUEUE_TYPE_PER_OBJECT, AddrId, Key, ErrorType),
                    Error
            end
    end;

handle_call({consume, ?QUEUE_ID_SYNC_BY_VNODE_ID, MessageBin}) ->
    case catch binary_to_term(MessageBin) of
        {'EXIT', Cause} ->
            {error, Cause};
        #sync_unit_of_vnode_message{vnode_id = ToVNodeId,
                                    node     = Node} ->
            case leo_redundant_manager_api:range_of_vnodes(ToVNodeId) of
                {ok, Res} ->
                    {ok, {CurRingHash, _PrevRingHash}} =
                        leo_redundant_manager_api:checksum(?CHECKSUM_RING),
                    ok = sync_vnodes(Node, CurRingHash, Res),
                    notify_rebalance_message_to_manager(ToVNodeId);
                Error ->
                    Error
            end
    end;

handle_call({consume, ?QUEUE_ID_REBALANCE, MessageBin}) ->
    case catch binary_to_term(MessageBin) of
        {'EXIT', Cause} ->
            {error, Cause};
        #rebalance_message{node     = Node,
                           vnode_id = VNodeId,
                           addr_id  = AddrId,
{'fraction_non_alphanumeric': 0.16062842446606285, 'fraction_numerical': 0.005479145700547914, 'mean_word_length': 1.4912256267409472, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 76, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 100, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2949513', 'n_tokens_mistral': 5226, 'n_tokens_neox': 5041, 'n_words': 951}