Merge branch 'main-v4.3' into using-erlang-system-time

CHANGES-4.3.md

@@ -17,6 +17,7 @@ File format:
 - Fix rule-engine update behaviour which may initialize actions for disabled rules. [#8849](https://github.com/emqx/emqx/pull/8849)
 - Fix JWT plugin don't support non-integer timestamp claims. [#8862](https://github.com/emqx/emqx/pull/8862)
 - Fix delayed publish inaccurate caused by os time change. [#8908](https://github.com/emqx/emqx/pull/8908)
+- Fix a possible dead loop caused by shared subscriptions with `shared_dispatch_ack_enabled=true`. [#8918](https://github.com/emqx/emqx/pull/8918)
 - Fix dashboard binding IP address not working. [#8916](https://github.com/emqx/emqx/pull/8916)
 
 

rebar.config

@@ -50,7 +50,7 @@
     , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.8.1.11"}}}
     , {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.5.2"}}}
     , {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v3.3.6"}}}
-    , {minirest, {git, "https://github.com/emqx/minirest", {tag, "0.3.7"}}}
+    , {minirest, {git, "https://github.com/emqx/minirest", {tag, "0.3.9"}}}
     , {ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.2"}}}
     , {replayq, {git, "https://github.com/emqx/replayq", {tag, "0.3.4"}}}
     , {pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {branch, "2.0.4"}}}

src/emqx.appup.src

@@ -2,18 +2,22 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
   [{"4.3.20",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]}]},
    {"4.3.19",
-    [{load_module,emqx_message,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
    {"4.3.18",
-    [{load_module,emqx_message,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
@@ -23,11 +27,11 @@
     [{load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx,brutal_purge,soft_purge,[]},
      {load_module,emqx_exclusive_subscription,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
@@ -94,12 +98,12 @@
      {load_module,emqx_packet,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_frame,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
@@ -506,6 +510,7 @@
      {load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
@@ -513,7 +518,6 @@
      {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
@@ -549,13 +553,13 @@
      {load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
      {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_packet,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
@@ -712,18 +716,22 @@
      {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
   [{"4.3.20",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]}]},
    {"4.3.19",
-    [{load_module,emqx_message,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
    {"4.3.18",
-    [{load_module,emqx_message,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
@@ -733,11 +741,11 @@
     [{load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx,brutal_purge,soft_purge,[]},
      {load_module,emqx_exclusive_subscription,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
@@ -807,12 +815,12 @@
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_frame,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
@@ -1193,6 +1201,7 @@
      {load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_metrics,brutal_purge,soft_purge,[]},
@@ -1204,7 +1213,6 @@
      {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
@@ -1234,6 +1242,7 @@
      {load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_metrics,brutal_purge,soft_purge,[]},
@@ -1244,7 +1253,6 @@
      {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_packet,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},

src/emqx_channel.erl

@@ -735,8 +735,8 @@ handle_deliver(Delivers, Channel = #channel{
     %% NOTE: Order is important here. While the takeover is in
     %% progress, the session cannot enqueue messages, since it already
     %% passed on the queue to the new connection in the session state.
-    NPendings = lists:append(Pendings,
+    NDelivers = ignore_local(ClientInfo, maybe_discard_shared_delivers(Delivers), ClientId, Session),
-        ignore_local(ClientInfo, maybe_nack(Delivers), ClientId, Session)),
+    NPendings = lists:append(Pendings, NDelivers),
     {ok, Channel#channel{pendings = NPendings}};
 
 handle_deliver(Delivers, Channel = #channel{
@@ -744,8 +744,8 @@ handle_deliver(Delivers, Channel = #channel{
         takeover   = false,
         session    = Session,
         clientinfo = #{clientid := ClientId} = ClientInfo}) ->
-    NSession = emqx_session:enqueue(ClientInfo,
+    NDelivers = ignore_local(ClientInfo, maybe_discard_shared_delivers(Delivers), ClientId, Session),
-        ignore_local(ClientInfo, maybe_nack(Delivers), ClientId, Session), Session),
+    NSession = emqx_session:enqueue(ClientInfo, NDelivers, Session),
     {ok, Channel#channel{session = NSession}};
 
 handle_deliver(Delivers, Channel = #channel{
@@ -776,12 +776,23 @@ ignore_local(ClientInfo, Delivers, Subscriber, Session) ->
                     end, Delivers).
 
 %% Nack delivers from shared subscription
-maybe_nack(Delivers) ->
+maybe_discard_shared_delivers(Delivers) ->
-    lists:filter(fun not_nacked/1, Delivers).
+    lists:filtermap(
-
+      fun({deliver, Topic, Msg}) ->
-not_nacked({deliver, _Topic, Msg}) ->
+              case emqx_shared_sub:is_ack_required(Msg) of
-    not (emqx_shared_sub:is_ack_required(Msg)
+                  false ->
-         andalso (ok == emqx_shared_sub:nack_no_connection(Msg))).
+                      true;
+                  true ->
+                      case emqx_shared_sub:is_retry_dispatch(Msg) of
+                          true ->
+                              %% force enqueue the retried shared deliver
+                              {true, {deliver, Topic, emqx_shared_sub:maybe_ack(Msg)}};
+                          false ->
+                              ok = emqx_shared_sub:nack_no_connection(Msg),
+                              false
+                      end
+              end
+      end, Delivers).
 
 %%--------------------------------------------------------------------
 %% Handle outgoing packet

src/emqx_shared_sub.erl

@@ -46,15 +46,14 @@
         , maybe_nack_dropped/1
         , nack_no_connection/1
         , is_ack_required/1
+        , is_retry_dispatch/1
         , get_group/1
         ]).
 
 %% for testing
 -ifdef(TEST).
--export([ subscribers/2
+-compile(export_all).
-        , ack_enabled/0
+-compile(nowarn_export_all).
-        , strategy/1
-        ]).
 -endif.
 
 %% gen_server callbacks
@@ -239,6 +238,13 @@ get_group(Msg) ->
 -spec(is_ack_required(emqx_types:message()) -> boolean()).
 is_ack_required(Msg) -> ?NO_ACK =/= get_group_ack(Msg).
 
+-spec(is_retry_dispatch(emqx_types:message()) -> boolean()).
+is_retry_dispatch(Msg) ->
+    case get_group_ack(Msg) of
+        {_Sender, {retry, _Group, _Ref}} -> true;
+        _ -> false
+    end.
+
 %% @doc Negative ack dropped message due to inflight window or message queue being full.
 -spec(maybe_nack_dropped(emqx_types:message()) -> store | drop).
 maybe_nack_dropped(Msg) ->
@@ -280,10 +286,15 @@ maybe_ack(Msg) ->
             Msg;
         Ack ->
             {Sender, Ref} = fetch_sender_ref(Ack),
-            Sender ! {Ref, ?ACK},
+            ack(Sender, Ref),
             without_group_ack(Msg)
     end.
 
+-spec(ack(pid(), reference()) -> ok).
+ack(Sender, Ref) ->
+    Sender ! {Ref, ?ACK},
+    ok.
+
 fetch_sender_ref({Sender, {_Type, _Group, Ref}}) -> {Sender, Ref};
 %% These clauses are for backward compatibility
 fetch_sender_ref({Sender, Ref}) -> {Sender, Ref}.

test/emqx_channel_SUITE.erl

@@ -481,6 +481,45 @@ t_handle_deliver_nl(_) ->
     NMsg = emqx_message:set_flag(nl, Msg),
     {ok, Channel} = emqx_channel:handle_deliver([{deliver, <<"t1">>, NMsg}], Channel).
 
+t_handle_deliver_shared_in_no_connection(_) ->
+    Grp = <<"g">>,
+    Sender = self(),
+    Ref1 = make_ref(),
+    Ref2 = make_ref(),
+    Chann = emqx_channel:set_field(conn_state, disconnected, channel()),
+
+    Msg0 = emqx_shared_sub:with_group_ack(
+             emqx_message:make(test, ?QOS_1, <<"t">>, <<"qos1">>),
+             Grp,
+             fresh,
+             Sender,
+             Ref1
+            ),
+    Msg1 = emqx_shared_sub:with_group_ack(
+             emqx_message:make(test, ?QOS_2, <<"t">>, <<"qos2">>),
+             Grp,
+             retry,
+             Sender,
+             Ref2
+            ),
+    Delivers = [{deliver, <<"+">>, Msg0}, {deliver, <<"+">>, Msg1}],
+
+    %% all shared msgs should be queued if shared_dispatch_ack_enabled=false
+    meck:new(emqx_shared_sub, [passthrough, no_history]),
+    meck:expect(emqx_shared_sub, is_ack_required, fun(_) -> false end),
+    {ok, Chann1} = emqx_channel:handle_deliver(Delivers, Chann),
+    ?assertEqual(2, proplists:get_value(mqueue_len, emqx_channel:stats(Chann1))),
+    meck:unload(emqx_shared_sub),
+
+    %% only fresh shared msgs should be queued if shared_dispatch_ack_enabled=true
+    meck:new(emqx_shared_sub, [passthrough, no_history]),
+    meck:expect(emqx_shared_sub, is_ack_required, fun(_) -> true end),
+    {ok, Chann2} = emqx_channel:handle_deliver(Delivers, Chann),
+    ?assertEqual(1, proplists:get_value(mqueue_len, emqx_channel:stats(Chann2))),
+    receive {Ref1, {shared_sub_nack, no_connection}} -> ok after 0 -> ?assert(false) end,
+    receive {Ref2, shared_sub_ack} -> ok after 0 -> ?assert(false) end,
+    meck:unload(emqx_shared_sub).
+
 %%--------------------------------------------------------------------
 %% Test cases for handle_out
 %%--------------------------------------------------------------------