feat(queue): fix static check issues

apps/emqx/priv/bpapi.versions

@@ -27,6 +27,7 @@
 {emqx_ds,2}.
 {emqx_ds,3}.
 {emqx_ds,4}.
+{emqx_ds_shared_sub,1}.
 {emqx_eviction_agent,1}.
 {emqx_eviction_agent,2}.
 {emqx_eviction_agent,3}.

apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl

@@ -364,7 +364,7 @@ agent_opts(#{session_id := SessionId}) ->
 now_ms() ->
     erlang:system_time(millisecond).
 
-is_use_finished(S, #srs{unsubscribed = Unsubscribed}) ->
+is_use_finished(_S, #srs{unsubscribed = Unsubscribed}) ->
     Unsubscribed.
 
 is_stream_fully_acked(S, SRS) ->

apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_null_agent.erl

@@ -14,7 +14,7 @@
     on_unsubscribe/2,
     on_stream_progress/2,
     on_info/2,
-    on_disconnect/1,
+    on_disconnect/2,
 
     renew_streams/1
 ]).
@@ -37,7 +37,7 @@ on_subscribe(_Agent, _TopicFilter, _SubOpts) ->
 on_unsubscribe(Agent, _TopicFilter) ->
     Agent.
 
-on_disconnect(Agent) ->
+on_disconnect(Agent, _) ->
     Agent.
 
 renew_streams(Agent) ->

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl

@@ -372,7 +372,7 @@ handle_leader_renew_stream_lease(GSM, VersionOld, VersionNew) ->
     %% Unexpected versions or state
     transition(GSM, ?connecting, #{}).
 
--spec handle_stream_progress(group_sm(), emqx_ds_shared_sub_proto:agent_stream_progress()) ->
+-spec handle_stream_progress(group_sm(), list(emqx_ds_shared_sub_proto:agent_stream_progress())) ->
     group_sm().
 handle_stream_progress(#{state := ?connecting} = GSM, _StreamProgresses) ->
     GSM;

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl

@@ -40,7 +40,7 @@
 -type agent_state() :: #{
     %% Our view of group gm's status
     %% it lags the actual state
-    state := emqx_ds_shared_sub_agent:status(),
+    state := ?waiting_replaying | ?replaying | ?waiting_updating | ?updating,
     prev_version := emqx_maybe:t(emqx_ds_shared_sub_proto:version()),
     version := emqx_ds_shared_sub_proto:version(),
     agent_metadata := emqx_ds_shared_sub_proto:agent_metadata(),

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_proto.erl

@@ -244,8 +244,9 @@ leader_invalidate(ToAgent, OfGroup) ->
 %% Internal API
 %%--------------------------------------------------------------------
 
-agent(_Id, Pid) ->
+agent(Id, Pid) ->
-    ?agent(_Id, Pid).
+    _ = Id,
+    ?agent(Id, Pid).
 
 format_streams(Streams) ->
     lists:map(

apps/emqx_ds_shared_sub/src/proto/emqx_ds_shared_sub_proto_v1.erl

@@ -14,6 +14,7 @@
     agent_connect_leader/5,
     agent_update_stream_states/5,
     agent_update_stream_states/6,
+    agent_disconnect/5,
 
     leader_lease_streams/6,
     leader_renew_stream_lease/4,
@@ -30,7 +31,7 @@ introduced_in() ->
     emqx_ds_shared_sub_proto:leader(),
     emqx_ds_shared_sub_proto:agent(),
     emqx_ds_shared_sub_proto:agent_metadata(),
-    emqx_ds_shared_sub_proto:topic_filter()
+    emqx_persistent_session_ds:share_topic_filter()
 ) -> ok.
 agent_connect_leader(Node, ToLeader, FromAgent, AgentMetadata, TopicFilter) ->
     erpc:cast(Node, emqx_ds_shared_sub_proto, agent_connect_leader, [