Message selectors for quorum queues

 ## What?

Add message selectors to quorum queues:
* with SQL syntax as per JMS spec
* with AMQP filter expressions, e.g. to allow for message groups

 ## How?

* message selector feature is enabled statically on the QQ via queue args
* hold a subset of message headers in memory as defined in the queue args
* Use gb_trees for high prio or normal prio messages instead of rabbit_fifo_q
* no consumer specific queues for messages
* each QQ consumer remembers its last scanned Ra index and continues
  scanning where it left off previously
* Use another gb_trees for prompt message expiry
* Add/recommend limits for
  * max consumers per queue
  * max messages per queue
  * max metadata size per message
  The latter two provide an overall limit on memory usage.

Notes:
* Round robin isn't guaranteed in all cases. However, this should be
  okay because JMS doesn't define how a broker should dispatch messages:
> "Apart from the requirements of any message selectors, Jakarta
Messaging does not define how messages are distributed between multiple
consumers on the same queue."
* gb_trees rebalancing should be acceptable given that emitting all Ra live
  indexes requires O(N) space anyway

Alternatives:
* List of lists approach was slower CPU wise due to scanning overhead,
  but better memory wise. Deletions must be stored separately in
  maps.
* Ideally we need a skip list instead of gb_trees because it's simple, avoids O(n)
  rebalance, and could even append in O(1). However, this cannot be implemented in
  Erlang. A NIF would be one solution, but isn't interoperable.
* Probabilistic skip lists are better suited than red-black trees.
  Red-black trees can't be efficiently implemnted in Erlang.

Next steps:
* end to end spike

Author: ansd

deps/rabbit/src/rabbit_fifo.erl

@@ -21,6 +21,12 @@
 
 -define(DELIVERY_SEND_MSG_OPTS, [local, ra_event]).
 
+%% These shorthands for section qualifiers are defined in filtex-v1.0-wd09 §6.4.4.4
+-define(HEADER_SECTION, h).
+-define(MESSAGE_ANNOTATIONS_SECTION, m).
+-define(PROPERTIES_SECTION, p).
+-define(APPLICATION_PROPERTIES_SECTION, a).
+
 -type optimised_tuple(A, B) :: nonempty_improper_list(A, B).
 
 -type option(T) :: undefined | T.
@@ -38,12 +44,33 @@
 %% in enqueue messages. Used to ensure ordering of messages send from the
 %% same process
 
+-type msg_section() ::
+header |
+message_annotations |
+properties |
+application_properties.
+
+%% AMQP §3.2
+-type msg_section_short() ::
+?HEADER_SECTION |
+?MESSAGE_ANNOTATIONS_SECTION |
+?PROPERTIES_SECTION |
+?APPLICATION_PROPERTIES_SECTION.
+
+-type field_name() :: atom() | binary().
+
+-type filter_fields() :: [{msg_section(), field_name()}].
+
+%% Message metadata held in memory for consumers to filter on.
+-type msg_metadata() :: [{msg_section_short(), field_name(), term()}].
+
 -type msg_header() :: msg_size() |
                       optimised_tuple(msg_size(), Expiry :: milliseconds()) |
                       #{size := msg_size(),
                         acquired_count => non_neg_integer(),
                         delivery_count => non_neg_integer(),
-                        expiry => milliseconds()}.
+                        expiry => milliseconds(),
+                        meta => msg_metadata()}.
 %% The message header:
 %% size: The size of the message payload in bytes.
 %% delivery_count: The number of unsuccessful delivery attempts.
@@ -116,6 +143,11 @@
 -define(DELIVERY_CHUNK_LIMIT_B, 128_000).
 
 -type milliseconds() :: non_neg_integer().
+-type timestamp() :: milliseconds().
+
+-type counter() :: non_neg_integer().
+-type return_counter() :: counter().
+
 -record(consumer_cfg,
         {meta = #{} :: consumer_meta(),
          pid :: pid(),
@@ -127,7 +159,9 @@
          %% command: `{credit, ReceiverDeliveryCount, Credit}'
          credit_mode :: credited | credit_mode(),
          lifetime = once :: once | auto,
-         priority = 0 :: integer()}).
+         priority = 0 :: integer(),
+         filter = [] :: rabbit_amqp_filtex:filter_expressions()
+        }).
 
 -record(consumer,
         {cfg = #consumer_cfg{},
@@ -138,7 +172,10 @@
          %% decremented for each delivery
          credit = 0 :: non_neg_integer(),
          %% AMQP 1.0 §2.6.7
-         delivery_count :: rabbit_queue_type:delivery_count()
+         delivery_count :: rabbit_queue_type:delivery_count(),
+         scanned_idxs = {0, 0} :: {HighPrio :: ra:index(),
+                                   NormalPrio :: ra:index()},
+         scanned_returns = 0 :: return_counter()
         }).
 
 -type consumer() :: #consumer{}.
@@ -169,26 +206,44 @@
          overflow_strategy = drop_head :: drop_head | reject_publish,
          max_length :: option(non_neg_integer()),
          max_bytes :: option(non_neg_integer()),
+         max_bytes_meta :: option(non_neg_integer()),
          %% whether single active consumer is on or not for this queue
          consumer_strategy = competing :: consumer_strategy(),
          %% the maximum number of unsuccessful delivery attempts permitted
          delivery_limit :: option(non_neg_integer()),
          expires :: option(milliseconds()),
          msg_ttl :: option(milliseconds()),
+         filter_enabled :: boolean(),
+         %% TODO Probably no need to store this filter here.
+         %% It should be good enough to store the 0.9.1 queue args filter table
+         %% in Khepri and to validate the consumer filter against this queue
+         %% filter in rabbit_quorum_queue when attaching the consumer.
+         %% Maybe still useful to expose effective filters in the Management UI?
+         filter_fields :: filter_fields(),
          unused_2 = ?NIL,
          unused_3 = ?NIL
         }).
 
 -record(rabbit_fifo,
         {cfg :: #cfg{},
          % unassigned messages
-         messages = rabbit_fifo_q:new() :: rabbit_fifo_q:state(),
+         messages = rabbit_fifo_q:new() :: rabbit_fifo_q:state() | rabbit_fifo_filter_q:state(),
          messages_total = 0 :: non_neg_integer(),
          % queue of returned msg_in_ids - when checking out it picks from
-         returns = lqueue:new() :: lqueue:lqueue(term()),
+         returns = lqueue:new() :: lqueue:lqueue(term()) | gb_trees:tree(
+                                                             return_counter(),
+                                                             msg()),
+         %% * only used if filtering is enabled
+         %% * contains messages that are available and have a TTL set
+         %%   (we do not expire acquired messages)
+         filter_msgs_expiry = gb_trees:empty() :: gb_trees:tree(
+                                                    {timestamp(), ra:index()},
+                                                    %% reference into where this msg is stored
+                                                    hi | no | return_counter()),
          % a counter of enqueues - used to trigger shadow copy points
          % reset to 0 when release_cursor gets stored
-         enqueue_count = 0 :: non_neg_integer(),
+         enqueue_count = 0 :: counter(),
+         return_count = 0 :: return_counter(),
          % a map containing all the live processes that have ever enqueued
          % a message to this queue
          enqueuers = #{} :: #{pid() => #enqueuer{}},
@@ -201,6 +256,11 @@
          unused_1 = ?NIL,
          % consumers need to reflect consumer state at time of snapshot
          consumers = #{} :: #{consumer_key() => consumer()},
+         %% TODO should only be used for filtering enabled queues.
+         %% All active consumers including consumers that may be down or may contain 0 credits.
+         %% An auxiliary data structure used as the base service queue after a new message has
+         %% been enqueued.
+         consumers_q = priority_queue:new() :: priority_queue:q(),
          % consumers that require further service are queued here
          service_queue = priority_queue:new() :: priority_queue:q(),
          %% state for at-least-once dead-lettering
@@ -223,8 +283,10 @@
                     checkpoint_max_indexes => non_neg_integer(),
                     max_length => non_neg_integer(),
                     max_bytes => non_neg_integer(),
+                    max_bytes_meta => non_neg_integer(),
                     overflow_strategy => drop_head | reject_publish,
                     single_active_consumer_on => boolean(),
+                    filter => false | filter_fields(),
                     delivery_limit => non_neg_integer() | -1,
                     expires => non_neg_integer(),
                     msg_ttl => non_neg_integer(),

deps/rabbit/src/rabbit_fifo.hrl

@@ -52,6 +52,7 @@
                    % status = up :: up | cancelled,
                    last_msg_id :: seq() | -1 | undefined,
                    ack = false :: boolean(),
+                   filter :: rabbit_amqp_filter:filter_expression(),
                    %% Remove this field when feature flag rabbitmq_4.0.0 becomes required.
                    delivery_count :: {credit_api_v1, rabbit_queue_type:delivery_count()} |
                                      credit_api_v2
@@ -165,7 +166,14 @@ enqueue(QName, Correlation, Msg,
                cfg = #cfg{soft_limit = SftLmt}} = State0) ->
     ServerId = pick_server(State0),
     % by default there is no correlation id
-    Cmd = rabbit_fifo:make_enqueue(self(), EnqueueSeq, Msg),
+    %% TODO Check which metadata this queue filters on.
+    {utf8, GroupId} = mc:property(group_id, Msg),
+    %% TODO rabbit_fifo should provide this shorthand.
+    %% Probably it's best if rabbit_fifo_filter provides some client API
+    %% to map the subset of metadata to filter on (set via queue args) to
+    %% how to extract this from mc and store this in e3{}.
+    MsgMeta = [{_ShorthandPropertiesSection = p, group_id, GroupId}],
+    Cmd = rabbit_fifo:make_enqueue(self(), EnqueueSeq, Msg, MsgMeta),
     ok = ra:pipeline_command(ServerId, Cmd, Seq, low),
     IsSlow = map_size(Pending) >= SftLmt,
     State = State0#state{pending = Pending#{Seq => {Correlation, Cmd}},
@@ -418,12 +426,15 @@ checkout(ConsumerTag, CreditMode, #{} = Meta,
                                 false -> {credit_api_v1, 0}
                             end,
             ConsumerKey = maps:get(key, Reply, ConsumerId),
+            Filter = maps:get(filter, Meta),
             SDels = maps:update_with(
                       ConsumerTag,
-                      fun (C) -> C#consumer{ack = Ack} end,
+                      fun (C) -> C#consumer{ack = Ack,
+                                            filter = Filter} end,
                       #consumer{key = ConsumerKey,
                                 last_msg_id = LastMsgId,
                                 ack = Ack,
+                                filter = Filter,
                                 delivery_count = DeliveryCount},
                       CDels0),
             {ok, Reply, State0#state{leader = Leader,

deps/rabbit/src/rabbit_fifo_client.erl

@@ -0,0 +1,31 @@
+-module(rabbit_fifo_filter).
+
+-include("rabbit_fifo.hrl").
+
+-export([filter/2]).
+
+filter(Filters, MsgMetaData) ->
+    %% "A message will pass through a filter-set if and only if
+    %% it passes through each of the named filters." [3.5.8]
+    lists:all(fun(Filter) ->
+                      filter0(Filter, MsgMetaData)
+              end, Filters).
+
+filter0({properties, KVList}, MsgMetaData) ->
+    %% "The filter evaluates to true if all properties enclosed in the filter expression
+    %% match the respective properties in the message."
+    %% [filtex-v1.0-wd09 4.2.4]
+    lists:all(
+      fun({RefField, RefVal}) ->
+              case lists:search(fun({?PROPERTIES_SECTION, Field, _Val})
+                                      when Field =:= RefField ->
+                                        true;
+                                   (_) ->
+                                        false
+                                end, MsgMetaData) of
+                  {value, {_, _, RefVal}} ->
+                      true;
+                  _ ->
+                      false
+              end
+      end, KVList).

deps/rabbit/src/rabbit_fifo_filter.erl

@@ -0,0 +1,156 @@
+%% A filterable queue with efficient random access and deletion.
+-module(rabbit_fifo_filter_q).
+
+-include("rabbit_fifo.hrl").
+-export([
+         new/0,
+         in/3,
+         size/1,
+         take/3,
+         take/4,
+         is_fully_scanned/2,
+         overview/1
+        ]).
+
+
+-record(?MODULE, {
+           hi :: gb_trees:tree(ra:index(), msg_header()),
+           no :: gb_trees:tree(ra:index(), msg_header()),
+           %% cache largest to avoid frequent gb_trees:largest/1 calls
+           hi_max :: ra:index(),
+           no_max :: ra:index()
+          }).
+
+-opaque state() :: #?MODULE{}.
+
+-export_type([state/0]).
+
+-spec new() -> state().
+new() ->
+    #?MODULE{hi = gb_trees:empty(),
+             no = gb_trees:empty(),
+             hi_max = 0,
+             no_max = 0}.
+
+-spec in(hi | no, msg(), state()) -> state().
+in(hi, ?MSG(RaIdx, Hdr), State = #?MODULE{hi = Hi,
+                                          hi_max = HiMax})
+  when RaIdx > HiMax ->
+    State#?MODULE{hi = gb_trees:insert(RaIdx, Hdr, Hi),
+                  hi_max = RaIdx};
+in(no, ?MSG(RaIdx, Hdr), State = #?MODULE{no = No,
+                                          no_max = NoMax})
+  when RaIdx > NoMax ->
+    State#?MODULE{no = gb_trees:insert(RaIdx, Hdr, No),
+                  no_max = RaIdx}.
+
+-spec size(state()) -> non_neg_integer().
+size(#?MODULE{hi = Hi,
+              no = No}) ->
+    gb_trees:size(Hi) +
+    gb_trees:size(No).
+
+%% Assumes the provided Ra index exists in the tree.
+-spec take(ra:index(), hi | no, state()) ->
+    {msg(), state()}.
+take(Idx, hi, State0 = #?MODULE{hi = Hi0,
+                                hi_max = HiMax0}) ->
+    {Hdr, Hi} = gb_trees:take(Idx, Hi0),
+    HiMax = new_largest(Idx, HiMax0, Hi),
+    State = State0#?MODULE{hi = Hi,
+                           hi_max = HiMax},
+    {?MSG(Idx, Hdr), State};
+take(Idx, no, State0 = #?MODULE{no = No0,
+                                no_max = NoMax0}) ->
+    {Hdr, No} = gb_trees:take(Idx, No0),
+    NoMax = new_largest(Idx, NoMax0, No),
+    State = State0#?MODULE{no = No,
+                           no_max = NoMax},
+    {?MSG(Idx, Hdr), State}.
+
+-spec take(timestamp(),
+           rabbit_amqp_filtex:filter_expressions(),
+           {ra:index(), ra:index()},
+           state()) ->
+    {empty, {ra:index(), ra:index()}} |
+    {msg(), {ra:index(), ra:index()}, state()}.
+take(RaTs, Filter, {HiIdx, NoIdx}, State0 = #?MODULE{hi = Hi0,
+                                                     hi_max = HiMax0})
+  when HiIdx < HiMax0 ->
+    case filter(RaTs, Filter, HiIdx, Hi0) of
+        none ->
+            take(RaTs, Filter, {HiMax0, NoIdx}, State0);
+        Msg = ?MSG(Idx, _Hdr) ->
+            Hi = gb_trees:delete(Idx, Hi0),
+            HiMax = new_largest(Idx, HiMax0, Hi),
+            State = State0#?MODULE{hi = Hi,
+                                   hi_max = HiMax},
+            {Msg, {Idx, NoIdx}, State}
+    end;
+take(RaTs, Filter, {HiIdx, NoIdx}, State0 = #?MODULE{no = No0,
+                                                     no_max = NoMax0})
+  when NoIdx < NoMax0 ->
+    case filter(RaTs, Filter, NoIdx, No0) of
+        none ->
+            {empty, {HiIdx, NoMax0}};
+        Msg = ?MSG(Idx, _Hdr) ->
+            No = gb_trees:delete(Idx, No0),
+            NoMax = new_largest(Idx, NoMax0, No),
+            State = State0#?MODULE{no = No,
+                                   no_max = NoMax},
+            {Msg, {HiIdx, Idx}, State}
+    end;
+take(_RaTs, _Filter, Idxs, _State) ->
+    {empty, Idxs}.
+
+new_largest(Idx, Idx, Tree) ->
+    %% The largest index was taken.
+    case gb_trees:is_empty(Tree) of
+        true ->
+            Idx;
+        false ->
+            {NewLargest, _Val} = gb_trees:largest(Tree),
+            NewLargest
+    end;
+new_largest(_IdxTaken, OldLargest, _Tree) ->
+    OldLargest.
+
+filter(RaTs, Filter, ScannedIdx, Tree) ->
+    FromIdx = ScannedIdx + 1,
+    Iter = gb_trees:iterator_from(FromIdx, Tree),
+    filter_msg0(RaTs, Filter, gb_trees:next(Iter)).
+
+filter_msg0(_RaTs, _Filter, none) ->
+    none;
+filter_msg0(RaTs, Filter, {Idx, Hdr = #{meta := Meta}, Iter}) ->
+    case rabbit_fifo:get_header(expiry, Hdr) of
+        ExpiryTs when is_integer(ExpiryTs) andalso RaTs >= ExpiryTs ->
+            %% Message expired.
+            filter_msg0(RaTs, Filter, gb_trees:next(Iter));
+        _ ->
+            case rabbit_fifo_filter:filter(Filter, Meta) of
+                true ->
+                    ?MSG(Idx, Hdr);
+                false ->
+                    filter_msg0(RaTs, Filter, gb_trees:next(Iter))
+            end
+    end.
+
+-spec is_fully_scanned({ra:index(), ra:index()}, state()) ->
+    boolean().
+is_fully_scanned({HiIdx, NoIdx}, #?MODULE{hi_max = HiMax,
+                                          no_max = NoMax}) ->
+    HiIdx >= HiMax andalso
+    NoIdx >= NoMax.
+
+-spec overview(state()) ->
+    #{len := non_neg_integer(),
+      num_hi := non_neg_integer(),
+      num_no := non_neg_integer()}.
+overview(#?MODULE{hi = Hi,
+                  no = No}) ->
+    NumHi = gb_trees:size(Hi),
+    NumNo = gb_trees:size(No),
+    #{len => NumHi + NumNo,
+      num_hi => NumHi,
+      num_no => NumNo}.