Osiris 1.6.6

This version contains a bit more error handling during replica start
to avoid logging process crashes during RabbitMQ shutdown.

rabbit_stream_queue: avoid double local pid query

When starting a new consumer. Also move some metrics
registration a bit later when we know we are likely to succeed.

(cherry picked from commit 19ab882)

# Conflicts:
#	deps/rabbit/src/rabbit_stream_queue.erl

Author: kjnilsson

MODULE.bazel

@@ -43,7 +43,7 @@ bazel_dep(
 
 bazel_dep(
     name = "rabbitmq_osiris",
-    version = "1.6.4",
+    version = "1.6.6",
     repo_name = "osiris",
 )
 

deps/rabbit/Makefile

@@ -147,7 +147,7 @@ TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client meck prop
 PLT_APPS += mnesia
 
 dep_syslog = git https://github.com/schlagert/syslog 4.0.0
-dep_osiris = git https://github.com/rabbitmq/osiris v1.6.4
+dep_osiris = git https://github.com/rabbitmq/osiris v1.6.6
 dep_systemd = hex 0.6.1
 dep_seshat = hex 0.4.0
 

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -239,6 +239,8 @@ writer_pid(StreamId) when is_list(StreamId) ->
     MFA = {?MODULE, query_writer_pid, [StreamId]},
     query_pid(StreamId, MFA).
 
+-spec local_pid(string()) ->
+    {ok, pid()} | {error, not_found | term()}.
 local_pid(StreamId) when is_list(StreamId) ->
     MFA = {?MODULE, query_local_pid, [StreamId, node()]},
     query_pid(StreamId, MFA).

deps/rabbit/src/rabbit_stream_queue.erl

@@ -219,10 +219,16 @@ consume(Q, #{no_ack := true}, _)
 consume(Q, #{limiter_active := true}, _State)
   when ?amqqueue_is_stream(Q) ->
     {error, global_qos_not_supported_for_queue_type};
+<<<<<<< HEAD
 consume(Q, Spec, QState0) when ?amqqueue_is_stream(Q) ->
+=======
+consume(Q, Spec,
+        #stream_client{filtering_supported = FilteringSupported} = QState0)
+  when ?amqqueue_is_stream(Q) ->
+>>>>>>> 19ab8824ce (Osiris 1.6.6)
     %% Messages should include the offset as a custom header.
-    case check_queue_exists_in_local_node(Q) of
-        ok ->
+    case get_local_pid(QState0) of
+        {LocalPid, QState} when is_pid(LocalPid) ->
             #{no_ack := NoAck,
               channel_pid := ChPid,
               prefetch_count := ConsumerPrefetchCount,
@@ -231,10 +237,12 @@ consume(Q, Spec, QState0) when ?amqqueue_is_stream(Q) ->
               args := Args,
               ok_msg := OkMsg} = Spec,
             QName = amqqueue:get_name(Q),
-            case parse_offset_arg(rabbit_misc:table_lookup(Args, <<"x-stream-offset">>)) of
+            case parse_offset_arg(
+                   rabbit_misc:table_lookup(Args, <<"x-stream-offset">>)) of
                 {error, _} = Err ->
                     Err;
                 {ok, OffsetSpec} ->
+<<<<<<< HEAD
                     _ = rabbit_stream_coordinator:register_local_member_listener(Q),
                     rabbit_core_metrics:consumer_created(ChPid, ConsumerTag, ExclusiveConsume,
                                                          not NoAck, QName,
@@ -245,9 +253,31 @@ consume(Q, Spec, QState0) when ?amqqueue_is_stream(Q) ->
                     %% do
                     maybe_send_reply(ChPid, OkMsg),
                     begin_stream(QState0, ConsumerTag, OffsetSpec, ConsumerPrefetchCount)
+=======
+                    FilterSpec = filter_spec(Args),
+                    case {FilterSpec, FilteringSupported} of
+                        {#{filter_spec := _}, false} ->
+                            {protocol_error, precondition_failed,
+                             "Filtering is not supported", []};
+                        _ ->
+                            rabbit_core_metrics:consumer_created(ChPid, ConsumerTag,
+                                                                 ExclusiveConsume,
+                                                                 not NoAck, QName,
+                                                                 ConsumerPrefetchCount,
+                                                                 false, up, Args),
+                            %% reply needs to be sent before the stream
+                            %% begins sending
+                            maybe_send_reply(ChPid, OkMsg),
+                            _ = rabbit_stream_coordinator:register_local_member_listener(Q),
+                            begin_stream(QState, ConsumerTag, OffsetSpec,
+                                         ConsumerPrefetchCount, FilterSpec)
+                    end
+>>>>>>> 19ab8824ce (Osiris 1.6.6)
             end;
-        Err ->
-            Err
+        {undefined, _} ->
+            {protocol_error, precondition_failed,
+             "queue '~ts' does not have a running replica on the local node",
+             [rabbit_misc:rs(amqqueue:get_name(Q))]}
     end.
 
 -spec parse_offset_arg(undefined |
@@ -284,8 +314,7 @@ get_local_pid(#stream_client{local_pid = Pid} = State)
 get_local_pid(#stream_client{leader = Pid} = State)
   when is_pid(Pid) andalso node(Pid) == node() ->
     {Pid, State#stream_client{local_pid = Pid}};
-get_local_pid(#stream_client{stream_id = StreamId,
-                             local_pid = undefined} = State) ->
+get_local_pid(#stream_client{stream_id = StreamId} = State) ->
     %% query local coordinator to get pid
     case rabbit_stream_coordinator:local_pid(StreamId) of
         {ok, Pid} ->
@@ -294,6 +323,7 @@ get_local_pid(#stream_client{stream_id = StreamId,
             {undefined, State}
     end.
 
+<<<<<<< HEAD
 begin_stream(#stream_client{name = QName, readers = Readers0} = State0,
              Tag, Offset, Max) ->
     {LocalPid, State} = get_local_pid(State0),
@@ -321,6 +351,32 @@ begin_stream(#stream_client{name = QName, readers = Readers0} = State0,
             {ok, State#stream_client{local_pid = LocalPid,
                                      readers = Readers0#{Tag => Str0}}}
     end.
+=======
+begin_stream(#stream_client{name = QName,
+                            readers = Readers0,
+                            local_pid = LocalPid} = State,
+             Tag, Offset, Max, Options)
+  when is_pid(LocalPid) ->
+    CounterSpec = {{?MODULE, QName, Tag, self()}, []},
+    {ok, Seg0} = osiris:init_reader(LocalPid, Offset, CounterSpec, Options),
+    NextOffset = osiris_log:next_offset(Seg0) - 1,
+    osiris:register_offset_listener(LocalPid, NextOffset),
+    StartOffset = case Offset of
+                      first -> NextOffset;
+                      last -> NextOffset;
+                      next -> NextOffset;
+                      {timestamp, _} -> NextOffset;
+                      _ -> Offset
+                  end,
+    Str0 = #stream{credit = Max,
+                   start_offset = StartOffset,
+                   listening_offset = NextOffset,
+                   log = Seg0,
+                   max = Max,
+                   reader_options = Options},
+    {ok, State#stream_client{local_pid = LocalPid,
+                             readers = Readers0#{Tag => Str0}}}.
+>>>>>>> 19ab8824ce (Osiris 1.6.6)
 
 cancel(_Q, ConsumerTag, OkMsg, ActingUser, #stream_client{readers = Readers0,
                                                           name = QName} = State) ->
@@ -340,8 +396,8 @@ cancel(_Q, ConsumerTag, OkMsg, ActingUser, #stream_client{readers = Readers0,
     end.
 
 credit(QName, CTag, Credit, Drain, #stream_client{readers = Readers0,
-                                           name = Name,
-                                           local_pid = LocalPid} = State) ->
+                                                  name = Name,
+                                                  local_pid = LocalPid} = State) ->
     {Readers1, Msgs} = case Readers0 of
                           #{CTag := #stream{credit = Credit0} = Str0} ->
                               Str1 = Str0#stream{credit = Credit0 + Credit},
@@ -355,7 +411,8 @@ credit(QName, CTag, Credit, Drain, #stream_client{readers = Readers0,
             true ->
                 case Readers1 of
                     #{CTag := #stream{credit = Credit1} = Str2} ->
-                        {Readers0#{CTag => Str2#stream{credit = 0}}, [{send_drained, {CTag, Credit1}}]};
+                        {Readers0#{CTag => Str2#stream{credit = 0}},
+                         [{send_drained, {CTag, Credit1}}]};
                     _ ->
                         {Readers1, []}
                 end;
@@ -376,6 +433,7 @@ deliver(QSs, #delivery{message = Msg, confirm = Confirm} = Delivery) ->
               {[{Q, S} | Qs], As ++ Actions}
       end, {[], []}, QSs).
 
+<<<<<<< HEAD
 deliver(_Confirm, #delivery{message = Msg, msg_seq_no = MsgId},
         #stream_client{name = Name,
                        leader = LeaderPid,
@@ -385,6 +443,20 @@ deliver(_Confirm, #delivery{message = Msg, msg_seq_no = MsgId},
                        soft_limit = SftLmt,
                        slow = Slow0} = State) ->
     ok = osiris:write(LeaderPid, WriterId, Seq, msg_to_iodata(Msg)),
+=======
+deliver0(MsgId, Msg,
+         #stream_client{name = Name,
+                        leader = LeaderPid,
+                        writer_id = WriterId,
+                        next_seq = Seq,
+                        correlation = Correlation0,
+                        soft_limit = SftLmt,
+                        slow = Slow0,
+                        filtering_supported = FilteringSupported} = State,
+         Actions0) ->
+    ok = osiris:write(LeaderPid, WriterId, Seq,
+                      stream_message(Msg, FilteringSupported)),
+>>>>>>> 19ab8824ce (Osiris 1.6.6)
     Correlation = case MsgId of
                       undefined ->
                           Correlation0;
@@ -923,17 +995,6 @@ stream_name(#resource{virtual_host = VHost, name = Name}) ->
 recover(Q) ->
     {ok, Q}.
 
-check_queue_exists_in_local_node(Q) ->
-    #{name := StreamId} = amqqueue:get_type_state(Q),
-    case rabbit_stream_coordinator:local_pid(StreamId) of
-        {ok, Pid} when is_pid(Pid) ->
-            ok;
-        _ ->
-            {protocol_error, precondition_failed,
-             "queue '~ts' does not have a replica on the local node",
-             [rabbit_misc:rs(amqqueue:get_name(Q))]}
-    end.
-
 maybe_send_reply(_ChPid, undefined) -> ok;
 maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg).