Repository: moquette-io/moquette
Pull request #739 (2023-03-20 14:46:24)

Implement session expiry cleanup for persisted sessions
## Release notes
Implement the logic to expire the not clean sessions.

## What does it do?
Track all expiration time after a session is closed or disconnects. Uses a delay queue to track the not clean session that expires and using a scheduled task, every second, check for the expired sessions and removes from the both SessionRegistry and also from the persistent store (SessionRepositoy).

## Why it's important for the user?
This feature let the user to auto-purge the state of  the not clean session after a certain amount of time configured at broker level in `moquette.conf` file through the `persistent_client_expiration` property which could have the format:
 `<number><unit>` such as 1d for 1 day. Admitted values to unit are `s` `m` `h` `d` `w` `M` `y` which respectively means seconds, minutes, hours, days, weeks, months and years. If not specified this global expiry defaults to INFINITE_EXPIRY which is ~80 years.

- close #662 

andsel: Hi @hylkevds may I ask you if you would like to be my buddy in this effort and provide a review to this functionality, please :pray: ?
hylkevds: While working on the subscription tree this weekend I noticed that subscriptions are not cleaned up when a session is cleaned up.
I noticed because the interceptor didn't get notified of any unsubscribe events when a (clean) session terminated due to a disconnect.

This may need its own issue, WDYT?
andsel: @hylkevds I think this could be a bug but:

> interceptor didn't get notified of any unsubscribe events

the interceptor is notified only when an UNSUBSCRIBE message is received.  When a clean session is wiped due to a disconnection it doesn't mean that the unsubscribe interceptor is notified, or do you thing it should?
andsel: @hylkevds please open an issue describing for that, describing the behavior. It will addressed in a separate PR.
andsel: Hi @hylkevds I've addressed all your latest concerns. Please could you review it?! :-)

Patch:
diff --git a/ChangeLog.txt b/ChangeLog.txt
index 4f270565..57fe9113 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,4 +1,5 @@
 Version 0.17-SNAPSHOT:
+   [feature] add saved session expiry configurable through the `persistent_client_expiration` setting (#739).
    [feature] implemented methods to forcibly disconnects a clients from a Server instance (#744).
    [refactory] moved code to handle session event loops from PostOffice into separate SessionEventLoopGroup (#742).
    [feature] added new callback method in interceptor to notify exceptions that happens on SessionEventLoop (#736).
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index d3e926d0..4b63aa9c 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -112,6 +112,8 @@ public final class BrokerConstants {
 
     public static final String STORAGE_CLASS_NAME = "storage_class";
 
+    public static final String PERSISTENT_CLEAN_EXPIRATION_PROPERTY_NAME = "persistent_client_expiration";
+
     public static final int FLIGHT_BEFORE_RESEND_MS = 5_000;
     public static final int INFLIGHT_WINDOW_SIZE = 10;
 
diff --git a/broker/src/main/java/io/moquette/broker/ISessionsRepository.java b/broker/src/main/java/io/moquette/broker/ISessionsRepository.java
index d172a2d2..640fc7f8 100644
--- a/broker/src/main/java/io/moquette/broker/ISessionsRepository.java
+++ b/broker/src/main/java/io/moquette/broker/ISessionsRepository.java
@@ -2,10 +2,13 @@ package io.moquette.broker;
 
 import io.netty.handler.codec.mqtt.MqttVersion;
 
+import java.time.Clock;
 import java.time.Instant;
 import java.util.Collection;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Used to store data about persisted sessions like MQTT version, session's properties.
@@ -13,29 +16,36 @@ import java.util.Optional;
 public interface ISessionsRepository {
 
     // Data class
-    final class SessionData {
+    final class SessionData implements Delayed {
         private final String clientId;
         private Instant expireAt = null;
         final MqttVersion version;
         private final int expiryInterval;
+        private transient final Clock clock;
 
         /**
          * Construct a new SessionData without expiration set yet.
+         *
+         * @expiryInterval seconds after which the persistent session could be dropped.
          * */
-        public SessionData(String clientId, MqttVersion version, int expiryInterval) {
+        public SessionData(String clientId, MqttVersion version, int expiryInterval, Clock clock) {
             this.clientId = clientId;
-            this.version = version;
+            this.clock = clock;
             this.expiryInterval = expiryInterval;
+            this.version = version;
         }
 
         /**
          * Construct SessionData with an expiration instant, created by loading from the storage.
+         *
+         * @expiryInterval seconds after which the persistent session could be dropped.
          * */
-        public SessionData(String clientId, Instant expireAt, MqttVersion version, int expiryInterval) {
-            this.expiryInterval = expiryInterval;
+        public SessionData(String clientId, Instant expireAt, MqttVersion version, int expiryInterval, Clock clock) {
             Objects.requireNonNull(expireAt, "An expiration time is requested");
+            this.clock = clock;
             this.clientId = clientId;
             this.expireAt = expireAt;
+            this.expiryInterval = expiryInterval;
             this.version = version;
         }
 
@@ -60,6 +70,11 @@ public interface ISessionsRepository {
             return expiryInterval;
         }
 
+        public SessionData withExpirationComputed() {
+            final Instant expiresAt = clock.instant().plusSeconds(expiryInterval);
+            return new SessionData(clientId, expiresAt, version, expiryInterval, clock);
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
@@ -82,6 +97,16 @@ public interface ISessionsRepository {
                 ", expiryInterval=" + expiryInterval +
                 '}';
         }
+
+        @Override
+        public long getDelay(TimeUnit unit) {
+            return unit.convert(expireAt.toEpochMilli() - clock.millis(), TimeUnit.MILLISECONDS);
+        }
+
+        @Override
+        public int compareTo(Delayed o) {
+            return Long.compare(getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
+        }
     }
 
     /**
@@ -93,4 +118,6 @@ public interface ISessionsRepository {
      * Save data composing a session, es MQTT version, creation date and properties but not queues or subscriptions.
      * */
     void saveSession(SessionData session);
+
+    void delete(SessionData session);
 }
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 3e789a42..42a2e1e9 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -55,6 +55,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.text.ParseException;
+import java.time.Clock;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -64,6 +65,7 @@ import java.util.UUID;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
+import static io.moquette.broker.Session.INFINITE_EXPIRY;
 import static io.moquette.logging.LoggingUtils.getInterceptorIds;
 
 public class Server {
@@ -210,6 +212,8 @@ public class Server {
             config.setProperty(BrokerConstants.DATA_PATH_PROPERTY_NAME, dataPath);
         }
 
+        final Clock clock = Clock.systemDefaultZone();
+
         if (Boolean.parseBoolean(config.getProperty(BrokerConstants.PERSISTENCE_ENABLED_PROPERTY_NAME))) {
             final Path dataPath = Paths.get(config.getProperty(BrokerConstants.DATA_PATH_PROPERTY_NAME));
             if (!dataPath.toFile().exists()) {
@@ -222,7 +226,7 @@ public class Server {
 
             LOG.debug("Configuring persistent subscriptions store and queues, path: {}", dataPath);
             final int autosaveInterval = Integer.parseInt(config.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30"));
-            h2Builder = new H2Builder(scheduler, dataPath, autosaveInterval).initStore();
+            h2Builder = new H2Builder(scheduler, dataPath, autosaveInterval, clock).initStore();
             queueRepository = initQueuesRepository(config, dataPath, h2Builder);
             LOG.trace("Configuring H2 subscriptions repository");
             subscriptionsRepository = h2Builder.subscriptionsRepository();
@@ -239,9 +243,18 @@ public class Server {
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         subscriptions.init(subscriptionsRepository);
         final Authorizator authorizator = new Authorizator(authorizatorPolicy);
-        sessions = new SessionRegistry(subscriptions, sessionsRepository, queueRepository, authorizator);
+
+        final int globalSessionExpiry;
+        if (config.getProperty(BrokerConstants.PERSISTENT_CLEAN_EXPIRATION_PROPERTY_NAME) != null) {
+            globalSessionExpiry = (int) config.durationProp(BrokerConstants.PERSISTENT_CLEAN_EXPIRATION_PROPERTY_NAME).toMillis() / 1000;
+        } else {
+            globalSessionExpiry = INFINITE_EXPIRY;
+        }
+
         final int sessionQueueSize = config.intProp(BrokerConstants.SESSION_QUEUE_SIZE, 1024);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(interceptor, sessionQueueSize);
+        sessions = new SessionRegistry(subscriptions, sessionsRepository, queueRepository, authorizator, scheduler,
+            clock, globalSessionExpiry, loopsGroup);
         dispatcher = new PostOffice(subscriptions, retainedRepository, sessions, interceptor, authorizator,
             loopsGroup);
         final BrokerConfiguration brokerConfig = new BrokerConfiguration(config);
diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java
index b20440ee..17829f6a 100644
--- a/broker/src/main/java/io/moquette/broker/Session.java
+++ b/broker/src/main/java/io/moquette/broker/Session.java
@@ -33,7 +33,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -51,9 +50,9 @@ import java.util.concurrent.atomic.AtomicReference;
 class Session {
 
     private static final Logger LOG = LoggerFactory.getLogger(Session.class);
-    // By specification session expiry value of 0xFFFFFFFF (UINT_MAX) (seconds) means
-    // session that doesn't expire, it's ~136 years, we can set a cap at 100 year
-    static final int INFINITE_EXPIRY = (int) Duration.ofDays(80 * 365).toMillis() / 1000;
+    // By specification session expiry value of 0xEFFFFFFF (UINT_MAX) (seconds) means
+    // session that doesn't expire, it's ~68 years.
+    static final int INFINITE_EXPIRY = Integer.MAX_VALUE;
 
     static class InFlightPacket implements Delayed {
 
@@ -502,6 +501,10 @@ class Session {
         }
     }
 
+    ISessionsRepository.SessionData getSessionData() {
+        return this.data;
+    }
+
     @Override
     public String toString() {
         return "Session{" +
diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
index 9e1eaaaf..2bf0af52 100644
--- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java
+++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
@@ -28,18 +28,32 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static io.moquette.broker.Session.INFINITE_EXPIRY;
 
 public class SessionRegistry {
 
+    private final ScheduledFuture<?> scheduledExpiredSessions;
+    private int globalExpirySeconds;
+    private final SessionEventLoopGroup loopsGroup;
+    static final Duration EXPIRED_SESSION_CLEANER_TASK_INTERVAL = Duration.ofSeconds(1);
+
     public abstract static class EnqueuedMessage {
 
         /**
@@ -122,18 +136,63 @@ public class SessionRegistry {
     private final ISessionsRepository sessionsRepository;
     private final IQueueRepository queueRepository;
     private final Authorizator authorizator;
+    private final DelayQueue<ISessionsRepository.SessionData> removableSessions = new DelayQueue<>();
+    private final Clock clock;
+
+    // Used in testing
+    SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory,
+                    ISessionsRepository sessionsRepository,
+                    IQueueRepository queueRepository,
+                    Authorizator authorizator,
+                    ScheduledExecutorService scheduler,
+                    SessionEventLoopGroup loopsGroup) {
+        this(subscriptionsDirectory, sessionsRepository, queueRepository, authorizator, scheduler, Clock.systemDefaultZone(), INFINITE_EXPIRY, loopsGroup);
+    }
 
     SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory,
                     ISessionsRepository sessionsRepository,
                     IQueueRepository queueRepository,
-                    Authorizator authorizator) {
+                    Authorizator authorizator,
+                    ScheduledExecutorService scheduler,
+                    Clock clock, int globalExpirySeconds,
+                    SessionEventLoopGroup loopsGroup) {
         this.subscriptionsDirectory = subscriptionsDirectory;
         this.sessionsRepository = sessionsRepository;
         this.queueRepository = queueRepository;
         this.authorizator = authorizator;
+        this.scheduledExpiredSessions = scheduler.scheduleWithFixedDelay(this::checkExpiredSessions,
+            EXPIRED_SESSION_CLEANER_TASK_INTERVAL.getSeconds(), EXPIRED_SESSION_CLEANER_TASK_INTERVAL.getSeconds(),
+            TimeUnit.SECONDS);
+        this.clock = clock;
+        this.globalExpirySeconds = globalExpirySeconds;
+        this.loopsGroup = loopsGroup;
         recreateSessionPool();
     }
 
+    private void checkExpiredSessions() {
+        List<ISessionsRepository.SessionData> expiredSessions = new ArrayList<>();
+        int drainedSessions = removableSessions.drainTo(expiredSessions);
+        LOG.debug("Retrieved {} expired sessions or {}", drainedSessions, removableSessions.size());
+        for (ISessionsRepository.SessionData expiredSession : expiredSessions) {
+            final String expiredAt = expiredSession.expireAt().map(Instant::toString).orElse("UNDEFINED");
+            LOG.debug("Removing session {}, expired on {}", expiredSession.clientId(), expiredAt);
+            remove(expiredSession.clientId());
+            sessionsRepository.delete(expiredSession);
+        }
+    }
+
+    private void trackForRemovalOnExpiration(ISessionsRepository.SessionData session) {
+        if (!session.expireAt().isPresent()) {
+            throw new RuntimeException("Can't track for expiration a session without expiry instant, client_id: " + session.clientId());
+        }
+        LOG.debug("start tracking the session {} for removal", session.clientId());
+        removableSessions.add(session);
+    }
+
+    private void untrackFromRemovalOnExpiration(ISessionsRepository.SessionData session) {
+        removableSessions.remove(session);
+    }
+
     private void recreateSessionPool() {
         final Set<String> queues = queueRepository.listQueueNames();
         for (ISessionsRepository.SessionData session : sessionsRepository.list()) {
@@ -143,10 +202,11 @@ public class SessionRegistry {
                 queues.remove(session.clientId());
                 Session rehydrated = new Session(session, false, persistentQueue);
                 pool.put(session.clientId(), rehydrated);
+                trackForRemovalOnExpiration(session);
             }
         }
         if (!queues.isEmpty()) {
-            LOG.error("Recreating sessions left {} unused queues. This is probably bug. Session IDs: {}", queues.size(), Arrays.toString(queues.toArray()));
+            LOG.error("Recreating sessions left {} unused queues. This is probably a bug. Session IDs: {}", queues.size(), Arrays.toString(queues.toArray()));
         }
     }
 
@@ -204,6 +264,8 @@ public class SessionRegistry {
             creationResult = new SessionCreationResult(oldSession, CreationModeEnum.REOPEN_EXISTING, true);
         }
 
+        untrackFromRemovalOnExpiration(creationResult.session.getSessionData());
+
         // case not covered new session is clean true/false and old session not in CONNECTED/DISCONNECTED
         return creationResult;
     }
@@ -236,10 +298,11 @@ public class SessionRegistry {
         } else {
             queue = new InMemoryQueue();
         }
-        // in MQTT3 cleanSession = true means  expiryInterval=0 else infinite
-        final int expiryInterval = clean ? 0 : INFINITE_EXPIRY;
+        // in MQTT3 cleanSession = true means expiryInterval=0 else infinite
+        final int expiryInterval = clean ? 0 : globalExpirySeconds;
+
         final ISessionsRepository.SessionData sessionData = new ISessionsRepository.SessionData(clientId,
-            MqttVersion.MQTT_3_1_1, expiryInterval);
+            MqttVersion.MQTT_3_1_1, expiryInterval, clock);
         if (msg.variableHeader().isWillFlag()) {
             final Session.Will will = createWill(msg);
             newSession = new Session(sessionData, clean, will, queue);
@@ -279,10 +342,9 @@ public class SessionRegistry {
         session.disconnect();
         if (session.expireImmediately()) {
             purgeSessionState(session);
-            return;
         } else {
-            // TODO if binded session has expiry, disconnect it and schedule a task to do the cleanup after that
-
+            //bound session has expiry, disconnect it and add to the queue for removal
+            trackForRemovalOnExpiration(session.getSessionData().withExpirationComputed());
         }
     }
 
@@ -300,7 +362,12 @@ public class SessionRegistry {
     void remove(String clientID) {
         final Session old = pool.remove(clientID);
         if (old != null) {
-            old.cleanUp();
+            // remove from expired tracker if present
+            removableSessions.remove(old.getSessionData());
+            loopsGroup.routeCommand(clientID, "Clean up removed session", () -> {
+                old.cleanUp();
+                return null;
+            });
         }
     }
 
@@ -351,6 +418,23 @@ public class SessionRegistry {
      * Close all resources related to session management
      */
     public void close() {
+        if (scheduledExpiredSessions.cancel(false)) {
+            LOG.info("Successfully cancelled expired sessions task");
+        } else {
+            LOG.warn("Can't cancel the execution of expired sessions task, was already cancelled? {}, was done? {}",
+                scheduledExpiredSessions.isCancelled(), scheduledExpiredSessions.isDone());
+        }
+        // Update all not clean session with the proper expiry date
+        updateNotCleanSessionsWithProperExpire();
         queueRepository.close();
     }
+
+    private void updateNotCleanSessionsWithProperExpire() {
+        pool.values().stream()
+            .filter(s -> !s.isClean()) // not clean session
+            .map(Session::getSessionData)
+            .filter(s -> !s.expireAt().isPresent()) // without expire set
+            .map(ISessionsRepository.SessionData::withExpirationComputed) // new SessionData with expireAt valued
+            .forEach(sessionsRepository::saveSession); // update the storage
+    }
 }
diff --git a/broker/src/main/java/io/moquette/broker/config/IConfig.java b/broker/src/main/java/io/moquette/broker/config/IConfig.java
index 0a0fcdff..ada8d3f6 100644
--- a/broker/src/main/java/io/moquette/broker/config/IConfig.java
+++ b/broker/src/main/java/io/moquette/broker/config/IConfig.java
@@ -18,6 +18,10 @@ package io.moquette.broker.config;
 
 import io.moquette.BrokerConstants;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalUnit;
+
 /**
  * Base interface for all configuration implementations (filesystem, memory or classpath)
  */
@@ -79,4 +83,37 @@ public abstract class IConfig {
         }
         return Boolean.parseBoolean(propertyValue);
     }
+
+    public Duration durationProp(String propertyName) {
+        String propertyValue = getProperty(propertyName);
+        final char timeSpecifier = propertyValue.charAt(propertyValue.length() - 1);
+        final TemporalUnit periodType;
+        switch (timeSpecifier) {
+            case 's':
+                periodType = ChronoUnit.SECONDS;
+                break;
+            case 'm':
+                periodType = ChronoUnit.MINUTES;
+                break;
+            case 'h':
+                periodType = ChronoUnit.HOURS;
+                break;
+            case 'd':
+                periodType = ChronoUnit.DAYS;
+                break;
+            case 'w':
+                periodType = ChronoUnit.WEEKS;
+                break;
+            case 'M':
+                periodType = ChronoUnit.MONTHS;
+                break;
+            case 'y':
+                periodType = ChronoUnit.YEARS;
+                break;
+            default:
+                throw new IllegalStateException("Can' parse duration property " + propertyName + " with value: " + propertyValue + ", admitted only h, d, w, m, y");
+
+        }
+        return Duration.of(Integer.parseInt(propertyValue.substring(0, propertyValue.length() - 1)), periodType);
+    }
 }
diff --git a/broker/src/main/java/io/moquette/persistence/H2Builder.java b/broker/src/main/java/io/moquette/persistence/H2Builder.java
index d2dd66d4..641c5cd6 100644
--- a/broker/src/main/java/io/moquette/persistence/H2Builder.java
+++ b/broker/src/main/java/io/moquette/persistence/H2Builder.java
@@ -12,6 +12,7 @@ import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.time.Clock;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
@@ -22,12 +23,14 @@ public class H2Builder {
     private final String storePath;
     private final int autosaveInterval; // in seconds
     private final ScheduledExecutorService scheduler;
+    private final Clock clock;
     private MVStore mvStore;
 
-    public H2Builder(ScheduledExecutorService scheduler, Path storePath, int autosaveInterval) {
+    public H2Builder(ScheduledExecutorService scheduler, Path storePath, int autosaveInterval, Clock clock) {
         this.storePath = storePath.resolve("moquette_store.h2").toAbsolutePath().toString();
         this.autosaveInterval = autosaveInterval;
         this.scheduler = scheduler;
+        this.clock = clock;
     }
 
     @SuppressWarnings("FutureReturnValueIgnored")
@@ -75,6 +78,6 @@ public class H2Builder {
     }
 
     public ISessionsRepository sessionsRepository() {
-        return new H2SessionsRepository(mvStore);
+        return new H2SessionsRepository(mvStore, clock);
     }
 }
diff --git a/broker/src/main/java/io/moquette/persistence/H2SessionsRepository.java b/broker/src/main/java/io/moquette/persistence/H2SessionsRepository.java
index e28e551b..e9667eca 100644
--- a/broker/src/main/java/io/moquette/persistence/H2SessionsRepository.java
+++ b/broker/src/main/java/io/moquette/persistence/H2SessionsRepository.java
@@ -9,6 +9,7 @@ import org.h2.mvstore.type.BasicDataType;
 import org.h2.mvstore.type.StringDataType;
 
 import java.nio.ByteBuffer;
+import java.time.Clock;
 import java.time.Instant;
 import java.util.Collection;
 
@@ -18,8 +19,10 @@ class H2SessionsRepository implements ISessionsRepository {
     private static final long UNDEFINED_INSTANT = -1;
 
     private final MVMap<String, SessionData> sessionMap;
+    private final Clock clock;
 
-    public H2SessionsRepository(MVStore mvStore) {
+    public H2SessionsRepository(MVStore mvStore, Clock clock) {
+        this.clock = clock;
         final MVMap.Builder<String, ISessionsRepository.SessionData> sessionTypeBuilder =
             new MVMap.Builder<String, ISessionsRepository.SessionData>()
                 .valueType(new SessionDataValueType());
@@ -37,6 +40,11 @@ class H2SessionsRepository implements ISessionsRepository {
         sessionMap.put(session.clientId(), session);
     }
 
+    @Override
+    public void delete(SessionData session) {
+        sessionMap.remove(session.clientId());
+    }
+
     /**
      * Codec data type to load and store SessionData instances
      */
@@ -70,9 +78,9 @@ class H2SessionsRepository implements ISessionsRepository {
             final int expiryInterval = buff.getInt();
 
             if (expiresAt == UNDEFINED_INSTANT) {
-                return new SessionData(clientId, version, expiryInterval);
+                return new SessionData(clientId, version, expiryInterval, clock);
             } else {
-                return new SessionData(clientId, Instant.ofEpochMilli(expiresAt), version, expiryInterval);
+                return new SessionData(clientId, Instant.ofEpochMilli(expiresAt), version, expiryInterval, clock);
             }
         }
 
diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionsRepository.java b/broker/src/main/java/io/moquette/persistence/MemorySessionsRepository.java
index bb2c11c0..25f0b04e 100644
--- a/broker/src/main/java/io/moquette/persistence/MemorySessionsRepository.java
+++ b/broker/src/main/java/io/moquette/persistence/MemorySessionsRepository.java
@@ -19,4 +19,9 @@ public class MemorySessionsRepository implements ISessionsRepository {
     public void saveSession(SessionData session) {
         sessions.put(session.clientId(), session);
     }
+
+    @Override
+    public void delete(SessionData session) {
+        sessions.remove(session.clientId());
+    }
 }
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index 40116ce0..e464f259 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/distribution/src/main/resources/moquette.conf
@@ -197,3 +197,17 @@ password_file config/password_file.conf
 # default: immediate
 #*********************************************************************
 # buffer_flush_millis immediate
+
+#*********************************************************************
+# Duration after which expire persisted sessions.
+#
+# persistent_client_expiration:
+#       This option allows the session of persistent clients (those with clean session set to false) that are not
+#       currently connected to be removed if they do not reconnect within a certain time frame.
+#       This is a non-standard option in MQTT v3.1. MQTT v3.1.1 and v5.0 allow brokers to remove client sessions.
+#       The expiration period should be an integer followed by one of s m h d w M y for seconds, minutes, hours, days, weeks,
+#       months and years respectively. For example: 2m or  14d or 1y
+# default: infinite expiry
+#*********************************************************************
+# persistent_client_expiration 3d
+


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/ForwardableClock.java b/broker/src/test/java/io/moquette/broker/ForwardableClock.java
new file mode 100644
index 00000000..50a2ee03
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/ForwardableClock.java
@@ -0,0 +1,38 @@
+package io.moquette.broker;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+
+/**
+ * Utility class to represent a clock that can be moved in time.
+ * This is used for tests that needs to verify conditions after a certain amount of time.
+ * */
+class ForwardableClock extends Clock {
+
+    private Clock currentClock;
+
+    ForwardableClock(Clock clock) {
+        this.currentClock = clock;
+    }
+
+    void forward(Duration period) {
+        currentClock = Clock.offset(currentClock, period);
+    }
+
+    @Override
+    public ZoneId getZone() {
+        return currentClock.getZone();
+    }
+
+    @Override
+    public Clock withZone(ZoneId zone) {
+        return currentClock.withZone(zone);
+    }
+
+    @Override
+    public Instant instant() {
+        return currentClock.instant();
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
index c0313aed..7df57d08 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
@@ -25,11 +25,14 @@ import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.MqttConnectMessage;
 import io.netty.handler.codec.mqtt.MqttMessageBuilders;
 import io.netty.handler.codec.mqtt.MqttVersion;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
 import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
@@ -58,9 +61,11 @@ public class MQTTConnectionConnectTest {
     private IAuthenticator mockAuthenticator;
     private PostOffice postOffice;
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
 
     @BeforeEach
     public void setUp() {
+        scheduler = Executors.newScheduledThreadPool(1);
         connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
 
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
@@ -72,8 +77,8 @@ public class MQTTConnectionConnectTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll, scheduler, loopsGroup);
         postOffice = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
                                     ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
 
@@ -81,6 +86,11 @@ public class MQTTConnectionConnectTest {
         channel = (EmbeddedChannel) sut.channel;
     }
 
+    @AfterEach
+    public void tearDown() {
+        scheduler.shutdown();
+    }
+
     private MQTTConnection createMQTTConnection(BrokerConfiguration config) {
         EmbeddedChannel channel = new EmbeddedChannel();
         return createMQTTConnection(config, channel, postOffice);
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
index b264bf50..3704d382 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
@@ -30,10 +30,13 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.netty.handler.codec.mqtt.MqttVersion;
 //import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 
 import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -52,6 +55,7 @@ public class MQTTConnectionPublishTest {
     private SessionRegistry sessionRegistry;
     private MqttMessageBuilders.ConnectBuilder connMsg;
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
 
     @BeforeEach
     public void setUp() {
@@ -59,9 +63,16 @@ public class MQTTConnectionPublishTest {
 
         BrokerConfiguration config = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
 
+        scheduler = Executors.newScheduledThreadPool(1);
+
         createMQTTConnection(config);
     }
 
+    @AfterEach
+    public void tearDown() {
+        scheduler.shutdown();
+    }
+
     private void createMQTTConnection(BrokerConfiguration config) {
         channel = new EmbeddedChannel();
         NettyUtils.clientID(channel, "test_client");
@@ -79,8 +90,8 @@ public class MQTTConnectionPublishTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll, scheduler, loopsGroup);
         final PostOffice postOffice = new PostOffice(subscriptions,
             new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
         return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, postOffice);
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
index 4aab271f..da47792e 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
@@ -25,11 +25,14 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.*;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
@@ -59,6 +62,7 @@ public class PostOfficeInternalPublishTest {
         new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryRetainedRepository retainedRepository;
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
 
     @BeforeEach
     public void setUp() throws ExecutionException, InterruptedException {
@@ -73,6 +77,11 @@ public class PostOfficeInternalPublishTest {
         ConnectionTestUtils.assertConnectAccepted(channel);
     }
 
+    @AfterEach
+    public void tearDown() {
+        scheduler.shutdown();
+    }
+
     private MQTTConnection createMQTTConnection(BrokerConfiguration config) {
         channel = new EmbeddedChannel();
         return createMQTTConnection(config, channel);
@@ -83,6 +92,7 @@ public class PostOfficeInternalPublishTest {
     }
 
     private void initPostOfficeAndSubsystems() {
+        scheduler = Executors.newScheduledThreadPool(1);
         subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
@@ -91,8 +101,8 @@ public class PostOfficeInternalPublishTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll, scheduler, loopsGroup);
         sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
     }
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
index 1c52b52e..4b95ee53 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
@@ -34,9 +34,7 @@ import org.junit.jupiter.api.Test;
 
 import java.nio.charset.Charset;
 import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
 import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
@@ -71,6 +69,7 @@ public class PostOfficePublishTest {
         new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryRetainedRepository retainedRepository;
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
 
     @BeforeEach
     public void setUp() {
@@ -85,6 +84,7 @@ public class PostOfficePublishTest {
 
     @AfterEach
     public void tearDown() {
+        scheduler.shutdown();
         sut.terminate();
     }
 
@@ -97,6 +97,7 @@ public class PostOfficePublishTest {
     }
 
     private void initPostOfficeAndSubsystems() {
+        scheduler = Executors.newScheduledThreadPool(1);
         subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
@@ -105,8 +106,8 @@ public class PostOfficePublishTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll, scheduler, loopsGroup);
         sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
     }
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
index 7a7d43fc..8e4d4134 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
@@ -28,15 +28,14 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.*;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.nio.charset.Charset;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
 import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
@@ -71,6 +70,7 @@ public class PostOfficeSubscribeTest {
     private SessionRegistry sessionRegistry;
     public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
 
     @BeforeEach
     public void setUp() {
@@ -82,12 +82,18 @@ public class PostOfficeSubscribeTest {
         createMQTTConnection(CONFIG);
     }
 
+    @AfterEach
+    public void tearDown() {
+        scheduler.shutdown();
+    }
+
     private void createMQTTConnection(BrokerConfiguration config) {
         channel = new EmbeddedChannel();
         connection = createMQTTConnection(config, channel);
     }
 
     private void prepareSUT() {
+        scheduler = Executors.newScheduledThreadPool(1);
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
 
         subscriptions = new CTrieSubscriptionDirectory();
@@ -97,8 +103,8 @@ public class PostOfficeSubscribeTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll, scheduler, loopsGroup);
         sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
     }
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
index 763af9ef..3e407d24 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
@@ -27,15 +27,14 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.*;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.nio.charset.Charset;
 import java.util.Collections;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
 import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
@@ -62,6 +61,7 @@ public class PostOfficeUnsubscribeTest {
     private SessionRegistry sessionRegistry;
     public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
 
     @BeforeEach
     public void setUp() {
@@ -73,12 +73,18 @@ public class PostOfficeUnsubscribeTest {
         createMQTTConnection(CONFIG);
     }
 
+    @AfterEach
+    public void tearDown() {
+        scheduler.shutdown();
+    }
+
     private void createMQTTConnection(BrokerConfiguration config) {
         channel = new EmbeddedChannel();
         connection = createMQTTConnection(config, channel);
     }
 
     private void prepareSUT() {
+        scheduler = Executors.newScheduledThreadPool(1);
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
 
         subscriptions = new CTrieSubscriptionDirectory();
@@ -88,8 +94,8 @@ public class PostOfficeUnsubscribeTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRegistry = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll, scheduler, loopsGroup);
         sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
     }
diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
index 286704b3..b6186c17 100644
--- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
@@ -31,14 +31,28 @@ import io.netty.handler.codec.mqtt.MqttConnectMessage;
 import io.netty.handler.codec.mqtt.MqttMessageBuilders;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.netty.handler.codec.mqtt.MqttVersion;
+import org.awaitility.Awaitility;
 import org.h2.mvstore.MVMap;
 import org.h2.mvstore.MVStore;
+import org.hamcrest.Matchers;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.nio.charset.StandardCharsets;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.Collection;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
 import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
@@ -54,9 +68,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 public class SessionRegistryTest {
 
     public static final boolean ANY_BOOLEAN = false;
+    private static final Logger LOG = LoggerFactory.getLogger(SessionRegistryTest.class);
     static final String FAKE_CLIENT_ID = "FAKE_123";
     static final String TEST_USER = "fakeuser";
     static final String TEST_PWD = "fakepwd";
+    public static final int GLOBAL_SESSION_EXPIRY_SECONDS = 100 * 24 * 60 * 60; // 100 days
 
     private MQTTConnection connection;
     private EmbeddedChannel channel;
@@ -65,14 +81,25 @@ public class SessionRegistryTest {
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID =
         new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryQueueRepository queueRepository;
+    private ScheduledExecutorService scheduler;
+    private final Clock pointInTimeFixedClock = Clock.fixed(Instant.parse("2023-03-26T18:09:30.00Z"), ZoneId.of("Europe/Rome"));
+    private ForwardableClock slidingClock = new ForwardableClock(pointInTimeFixedClock);
+    private ISessionsRepository sessionRepository;
 
     @BeforeEach
     public void setUp() {
         connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
 
+        scheduler = Executors.newScheduledThreadPool(1);
+
         createMQTTConnection(ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID);
     }
 
+    @AfterEach
+    public void tearDown() {
+        scheduler.shutdown();
+    }
+
     private void createMQTTConnection(BrokerConfiguration config) {
         channel = new EmbeddedChannel();
         connection = createMQTTConnection(config, channel);
@@ -89,8 +116,9 @@ public class SessionRegistryTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        sut = new SessionRegistry(subscriptions, memorySessionsRepository(), queueRepository, permitAll);
         final SessionEventLoopGroup loopsGroup = new SessionEventLoopGroup(ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, 1024);
+        sessionRepository = memorySessionsRepository();
+        sut = new SessionRegistry(subscriptions, sessionRepository, queueRepository, permitAll, scheduler, slidingClock, GLOBAL_SESSION_EXPIRY_SECONDS, loopsGroup);
         final PostOffice postOffice = new PostOffice(subscriptions,
             new MemoryRetainedRepository(), sut, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, loopsGroup);
         return new MQTTConnection(channel, config, mockAuthenticator, sut, postOffice);
@@ -98,7 +126,7 @@ public class SessionRegistryTest {
 
     @Test
     public void testConnAckContainsSessionPresentFlag() {
-        System.out.println("testConnAckContainsSessionPresentFlag invoked");
+        LOG.info("testConnAckContainsSessionPresentFlag invoked");
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
                                         .protocolVersion(MqttVersion.MQTT_3_1_1)
                                         .build();
@@ -123,6 +151,7 @@ public class SessionRegistryTest {
 
     @Test
     public void connectWithCleanSessionUpdateClientSession() throws ExecutionException, InterruptedException {
+        LOG.info("connectWithCleanSessionUpdateClientSession");
         // first connect with clean session true
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
         connection.processConnect(msg).completableFuture().get();
@@ -190,6 +219,7 @@ public class SessionRegistryTest {
 
     @Test
     public void testSerializabilityOfPublishedMessage() {
+        LOG.info("testSerializabilityOfPublishedMessage");
         MVStore mvStore = new MVStore.Builder()
             .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH)
             .autoCommitDisabled()
@@ -227,4 +257,67 @@ public class SessionRegistryTest {
             assertFalse(dbFile.exists());
         }
     }
+
+    @Test
+    public void givenSessionWithExpireTimeWhenAfterExpirationIsPassedThenSessionIsRemoved() {
+        LOG.info("givenSessionWithExpireTimeWhenAfterExpirationIsPassedThenSessionIsRemoved");
+
+        // insert a not clean session that should expire in GLOBAL_SESSION_EXPIRY_SECONDS (100 days)
+        final String clientId = "client_to_be_removed";
+        final SessionRegistry.SessionCreationResult res = sut.createOrReopenSession(connMsg.cleanSession(false).build(), clientId, "User");
+        assertEquals(SessionRegistry.CreationModeEnum.CREATED_CLEAN_NEW, res.mode, "Not clean session must be created");
+
+        // remove it, so that it's tracked in the inner delay queue
+        sut.connectionClosed(res.session);
+        assertEquals(1, sessionRepository.list().size(), "Not clean session must be persisted");
+
+        // move time forward
+        Duration moreThenSessionExpiration = Duration.ofSeconds(GLOBAL_SESSION_EXPIRY_SECONDS).plusSeconds(10);
+        slidingClock.forward(moreThenSessionExpiration);
+
+        // check the session has been removed
+        Awaitility
+            .await()
+            .atMost(3 * SessionRegistry.EXPIRED_SESSION_CLEANER_TASK_INTERVAL.toMillis(), TimeUnit.MILLISECONDS)
+            .until(sessionsList(), Matchers.empty());
+    }
+
+    @Test
+    public void givenSessionThatExpiresWhenReopenIsNotAnymoreTrackedForExpiration() throws InterruptedException {
+        LOG.info("givenSessionThatExpiresWhenReopenIsNotAnymoreTrackedForExpiration");
+        final String clientId = "client_to_be_removed";
+        SessionRegistry.SessionCreationResult res = sut.createOrReopenSession(connMsg.cleanSession(false).build(), clientId, "User");
+        assertEquals(SessionRegistry.CreationModeEnum.CREATED_CLEAN_NEW, res.mode, "Non clean session must be created");
+        res.session.completeConnection();
+
+        // remove it, so that it's tracked in the inner delay queue
+        sut.connectionClosed(res.session);
+        assertEquals(1, sessionRepository.list().size(), "Non clean session must be persisted");
+
+        // Exercise
+        // reopen the session
+        res = sut.createOrReopenSession(connMsg.cleanSession(false).build(), clientId, "User");
+        assertEquals(SessionRegistry.CreationModeEnum.REOPEN_EXISTING, res.mode, "Non clean session must be re-opened");
+
+        // move time forward
+        Duration moreThenSessionExpiration = Duration.ofSeconds(GLOBAL_SESSION_EXPIRY_SECONDS).plusSeconds(10);
+        slidingClock.forward(moreThenSessionExpiration);
+
+        // Verify that the session reopened is still listed
+        final Collection<ISessionsRepository.SessionData> activeSessions = sessionRepository.list();
+        assertEquals(1, activeSessions.size(), "There must be active one session");
+        final ISessionsRepository.SessionData element = activeSessions.iterator().next();
+        assertFalse(element.expireAt().isPresent(), "Shouldn't have an expiration configured");
+
+        // wait the session expiry thread kicks in for at least one execution
+        Thread.sleep(3 * SessionRegistry.EXPIRED_SESSION_CLEANER_TASK_INTERVAL.toMillis());
+        Awaitility
+            .await()
+            .atMost(2 * SessionRegistry.EXPIRED_SESSION_CLEANER_TASK_INTERVAL.toMillis(), TimeUnit.MILLISECONDS)
+            .until(sessionsList(), Matchers.not(Matchers.empty()));
+    }
+
+    private Callable<Collection<ISessionsRepository.SessionData>> sessionsList() {
+        return () -> sessionRepository.list();
+    }
 }
diff --git a/broker/src/test/java/io/moquette/broker/SessionTest.java b/broker/src/test/java/io/moquette/broker/SessionTest.java
index 00fb2b63..66c6c237 100644
--- a/broker/src/test/java/io/moquette/broker/SessionTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionTest.java
@@ -12,6 +12,8 @@ import org.junit.jupiter.api.Test;
 
 import static io.moquette.BrokerConstants.FLIGHT_BEFORE_RESEND_MS;
 import io.moquette.broker.subscriptions.Subscription;
+
+import java.time.Clock;
 import java.util.Arrays;
 import org.assertj.core.api.Assertions;
 
@@ -33,7 +35,8 @@ public class SessionTest {
     public void setUp() {
         testChannel = new EmbeddedChannel();
         queuedMessages = new InMemoryQueue();
-        final ISessionsRepository.SessionData data = new ISessionsRepository.SessionData(CLIENT_ID, MqttVersion.MQTT_3_1_1, INFINITE_EXPIRY);
+        final Clock clock = Clock.systemDefaultZone();
+        final ISessionsRepository.SessionData data = new ISessionsRepository.SessionData(CLIENT_ID, MqttVersion.MQTT_3_1_1, INFINITE_EXPIRY, clock);
         client = new Session(data, true, null, queuedMessages);
         createConnection(client);
     }