skip putting already sent messages into staging buffer to avoid session memory leaking

Author: popduke

bifromq-inbox/bifromq-inbox-client/src/main/java/com/baidu/bifromq/inbox/client/InboxReader.java

@@ -24,10 +24,8 @@ class InboxReader implements IInboxClient.IInboxReader {
     private final long incarnation;
     private final InboxFetchPipeline ppln;
     private final long sessionId = HLC.INST.get();
-    private int latestBufferCapacity = 100;
     private volatile long lastFetchQoS0Seq = -1;
     private volatile long lastFetchSendBufferSeq = -1;
-    private volatile long lastFetchQoS2Seq = -1;
 
     public InboxReader(String inboxId,
                        long incarnation,
@@ -54,7 +52,6 @@ public void fetch(Consumer<Fetched> consumer) {
 
     @Override
     public void hint(int bufferCapacity) {
-        latestBufferCapacity = bufferCapacity;
         try {
             ppln.hint(sessionId, inboxId, incarnation, bufferCapacity, lastFetchQoS0Seq, lastFetchSendBufferSeq);
         } catch (Throwable e) {

bifromq-mqtt/bifromq-mqtt-server/src/main/java/com/baidu/bifromq/mqtt/handler/MQTTPersistentSessionHandler.java

@@ -195,7 +195,9 @@ public final void channelInactive(ChannelHandlerContext ctx) {
         memUsage.addAndGet(-estBaseMemSize());
         int remainInboxSize =
             stagingBuffer.values().stream().reduce(0, (acc, msg) -> acc + msg.estBytes(), Integer::sum);
-        memUsage.addAndGet(-remainInboxSize);
+        if (remainInboxSize > 0) {
+            memUsage.addAndGet(-remainInboxSize);
+        }
         ctx.fireChannelInactive();
     }
 
@@ -502,6 +504,10 @@ private void pubQoS0Message(InboxMessage inboxMsg) {
     }
 
     private void pubBufferedMessage(InboxMessage inboxMsg) {
+        if (inboxMsg.getSeq() < nextSendSeq) {
+            // do not buffer message that has been sent
+            return;
+        }
         String topicFilter = inboxMsg.getTopicFilter();
         TopicFilterOption option = inboxMsg.getOption();
         addFgTask(authProvider.checkPermission(clientInfo(), buildSubAction(topicFilter, option.getQos())))
@@ -515,8 +521,9 @@ private void pubBufferedMessage(InboxMessage inboxMsg) {
                 tenantMeter.timer(msg.qos() == AT_LEAST_ONCE ? MqttQoS1InternalLatency : MqttQoS2InternalLatency)
                     .record(HLC.INST.getPhysical() - message.getTimestamp(), TimeUnit.MILLISECONDS);
                 SubMessage prev = stagingBuffer.put(inboxMsg.getSeq(), msg);
-                assert prev == null;
-                memUsage.addAndGet(msg.estBytes());
+                if (prev == null) {
+                    memUsage.addAndGet(msg.estBytes());
+                }
             });
     }