From 8dd4d88d5b1a75b63ccf7778a47a78adee2c6e4c Mon Sep 17 00:00:00 2001
From: lafirest <blankalupo@163.com>
Date: Fri, 26 Nov 2021 14:57:25 +0800
Subject: [PATCH] fix(emx_slow_updates): fix the error of topk update (#6312)

---
 apps/emqx_plugin_libs/src/emqx_slow_subs/emqx_slow_subs.erl | 6 ++++--
 src/emqx_slow_subs/emqx_message_latency_stats.erl           | 5 +++--
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/apps/emqx_plugin_libs/src/emqx_slow_subs/emqx_slow_subs.erl b/apps/emqx_plugin_libs/src/emqx_slow_subs/emqx_slow_subs.erl
index a50b5eb0e..e96c2a907 100644
--- a/apps/emqx_plugin_libs/src/emqx_slow_subs/emqx_slow_subs.erl
+++ b/apps/emqx_plugin_libs/src/emqx_slow_subs/emqx_slow_subs.erl
@@ -106,8 +106,10 @@ on_stats_update(#{clientid := ClientId,
     %% check whether the client is in the table
     case ets:lookup(?TOPK_TAB, LastIndex) of
         [#top_k{index = Index}] ->
-            %% if last value == the new value, return
-            true;
+            %% if last value == the new value, update the type and last_update_time
+            %% XXX for clients whose latency are stable for a long time, is it possible to reduce updates?
+            ets:insert(?TOPK_TAB,
+                       #top_k{index = Index, type = Type, last_update_time = Ts});
         [_] ->
             %% if Latency > minimum value, we should update it
             %% if Latency < minimum value, maybe it can replace the minimum value
diff --git a/src/emqx_slow_subs/emqx_message_latency_stats.erl b/src/emqx_slow_subs/emqx_message_latency_stats.erl
index f9661ab91..dfeba1c68 100644
--- a/src/emqx_slow_subs/emqx_message_latency_stats.erl
+++ b/src/emqx_slow_subs/emqx_message_latency_stats.erl
@@ -95,11 +95,12 @@ call_hook(_, _, _, Latency, #{threshold := Threshold} = S)
     S;
 
 call_hook(ClientId, Now, Type, Latency, #{last_insert_value := LIV} = Stats) ->
+    ToInsert = erlang:floor(Latency),
     Arg = #{clientid => ClientId,
-            latency => erlang:floor(Latency),
+            latency => ToInsert,
             type => Type,
             last_insert_value => LIV,
             update_time => Now},
     emqx:run_hook('message.slow_subs_stats', [Arg]),
-    Stats#{last_insert_value := Latency,
+    Stats#{last_insert_value := ToInsert,
            last_access_time := Now}.