Fixed SegmentedQueues not being cleaned up on session purge by hylkevds · Pull Request #833 · moquette-io/moquette

hylkevds · 2024-04-29T14:37:33Z

Segmented queues were closed, but not cleared on session close. This resulted in segments not being returned to the pool, and thus disk-space not being released. The queues were also not removed from the QueuePool, resulting in a memory leak.

Relates #734

andsel

Hi @hylkevds thanks for this PR. I would also add unit test on the unsafequeues.Queue or unsafequeues.QueuePool to verify that a queue purge or close give back the segments to be reused.

andsel · 2024-05-24T13:58:29Z

broker/src/main/java/io/moquette/broker/SessionRegistry.java

    void remove(String clientID) {
        final Session old = pool.remove(clientID);
        if (old != null) {
+            unsubscribe(old);


Are we sure that pool's session is the same instance of current session?

The only situation where it could happen that there are two sessions with the same clientID is when a client re-connects with cleanSession=true. In that case the old session is purged before the new session is created.

I can change the code to only remove from the pool if the session is the same as the one in the pool.

That just lead me to a different potential issue, with the sessionExpirationService timer. I think that timer could trigger right as a client is reconnecting, and as it is based on the clientID, it could kill a session that was just made. Shouldn't removeExpiredSession check if the session really is expired, and not reconnected? And probably do the cleanup on the session thread?

I think that timer could trigger right as a client is reconnecting, and as it is based on the clientID, it could kill a session that was just made. Shouldn't removeExpiredSession check if the session really is expired, and not reconnected? And probably do the cleanup on the session thread?

Why do you think that that case could happen? Looking at the code, when a session disconnect it's also untracked from the session expiration service. When it reconnects it's tracked again, reinserting the reference into the pool managed by the session expiration service.
In this way all session tracked by the expiration service are session present in the session registry, and so can't be the case of a duplication.

The session expiration service only contains disconnected sessions that have cleanSession=false. Sessions that are connected don't need to be tracked since they can't expire, and sessions with cleanSession=true are cleaned up on disconnect.

Imagine a disconnected session with cleanSession=false. The expiration service finds it in the queue and removes it for cleaning, but there is a time interval between it being removed from the queue and it being actually cleaned. Either because there is a large batch of sessions to expire, or because the thread is put in hold. In that time the user reconnects, and a new session is made that inherits the subscriptions and clientID of the old session. Now the expiration thread is reactivated and kills the thread with the stored clientID...

@hylkevds you are right! Very good explanation of the case. I think this aspect should be kept out of this PR, and an issue tracking the problem should opened so be solved. Thanks for pointing it out 👍

hylkevds · 2024-05-25T19:17:47Z

Hi @hylkevds thanks for this PR. I would also add unit test on the unsafequeues.Queue or unsafequeues.QueuePool to verify that a queue purge or close give back the segments to be reused.

I'll look into making some tests for this.

hylkevds · 2024-08-03T17:22:53Z

@andsel: I added a test to verify the queue cleanup and page-file re-use.

andsel

Hi @hylkevds the changes are almost good, just a couple of nitpicks on the log messages, then the PR is done.

broker/src/main/java/io/moquette/broker/unsafequeues/QueuePool.java

andsel · 2024-08-17T14:54:06Z

broker/src/main/java/io/moquette/broker/unsafequeues/QueuePool.java

+        segmentsAllocationLock.lock();
+        try {
+            while (segmentRef != null) {
+                LOG.debug("Consumed tail segment {} from queue {}", segmentRef, queueName);


Would rewrite this log as

Suggested change

LOG.debug("Consumed tail segment {} from queue {}", segmentRef, queueName);

LOG.debug("Purging segment {} from queue {}", segmentRef, queueName);

No problem, I added both logging lines.

andsel

Please keep this PR with only changes related to the persistent queues, that would help a lot the review and the change history of the project.
Please create a follow up PR for the issue related to session management.

andsel · 2025-07-26T13:59:54Z

broker/src/main/java/io/moquette/broker/SessionRegistry.java

-    void remove(String clientID) {
-        final Session old = pool.remove(clientID);
-        if (old != null) {
+    void remove(Session session) {


I would keep the fixes for session management out of this PR, so that's more clear what this PR does and simplify to understand the change history, reducing the scope of the PR

I've split the commit containing the SessionRegistry changes into a separate branch. PR coming soon...

Segmented queues were closed, but not cleared on session close. This resulted in segments not being returned to the pool, and thus disk-space not being released. The queues were also not removed from the QueuePool, resulting in a memory leak.

andsel

LGTM

andsel · 2025-08-02T10:53:43Z

Thanks @hylkevds for your contribution to the project!

hylkevds requested a review from andsel April 30, 2024 06:49

andsel requested changes May 24, 2024

View reviewed changes

hylkevds force-pushed the fix_segmentedQueueCleanup branch from c22869c to c20f3c5 Compare May 25, 2024 19:19

hylkevds force-pushed the fix_segmentedQueueCleanup branch from c20f3c5 to 9acaf3a Compare August 3, 2024 17:22

andsel requested changes Aug 17, 2024

View reviewed changes

hylkevds force-pushed the fix_segmentedQueueCleanup branch from d88b42f to e776a3a Compare June 17, 2025 11:03

hylkevds requested a review from andsel June 17, 2025 11:04

hylkevds self-assigned this Jun 17, 2025

andsel requested changes Jul 26, 2025

View reviewed changes

hylkevds force-pushed the fix_segmentedQueueCleanup branch from e776a3a to a1d9cc4 Compare July 27, 2025 19:01

hylkevds added 3 commits July 27, 2025 21:31

Added test for segment cleanup and page file re-use.

7c092d2

Added debug messages about queue purge

0c8e934

hylkevds force-pushed the fix_segmentedQueueCleanup branch from a1d9cc4 to 0c8e934 Compare July 27, 2025 19:34

andsel approved these changes Aug 2, 2025

View reviewed changes

andsel merged commit ebd67ad into moquette-io:main Aug 2, 2025
3 of 4 checks passed

hylkevds deleted the fix_segmentedQueueCleanup branch August 25, 2025 12:32

	LOG.debug("Consumed tail segment {} from queue {}", segmentRef, queueName);
	LOG.debug("Purging segment {} from queue {}", segmentRef, queueName);

Conversation

hylkevds commented Apr 29, 2024

Uh oh!

andsel left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andsel Jun 2, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hylkevds commented May 25, 2024

Uh oh!

hylkevds commented Aug 3, 2024

Uh oh!

andsel left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andsel left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andsel left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

andsel commented Aug 2, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

andsel Jun 2, 2024 •

edited

Loading