Fix for unlimited maximum message expiry interval

README-CN.md

@@ -183,7 +183,7 @@ server := mqtt.New(&mqtt.Options{
 
 关于决定默认配置的值，在这里进行一些说明：
 
-- 默认情况下，server.Options.Capabilities.MaximumMessageExpiryInterval 的值被设置为 86400（24小时），以防止在使用默认配置时网络上暴露服务器而受到恶意DOS攻击（如果不配置到期时间将允许无限数量的保留retained/待发送inflight消息累积）。如果您在一个受信任的环境中运行，或者您有更大的保留期容量，您可以选择覆盖此设置（设置为 0 或 math.MaxInt 以取消到期限制）。
+- 默认情况下，server.Options.Capabilities.MaximumMessageExpiryInterval 的值被设置为 86400（24小时），以防止在使用默认配置时网络上暴露服务器而受到恶意DOS攻击（如果不配置到期时间将允许无限数量的保留retained/待发送inflight消息累积）。如果您在一个受信任的环境中运行，或者您有更大的保留期容量，您可以选择覆盖此设置（设置为 0 或 math.MaxInt64 以取消到期限制）。
 
 ## 事件钩子(Event Hooks)
 

README.md

@@ -184,7 +184,7 @@ Review the mqtt.Options, mqtt.Capabilities, and mqtt.Compatibilities structs for
 
 Some choices were made when deciding the default configuration that need to be mentioned here:
 
-- By default, the value of `server.Options.Capabilities.MaximumMessageExpiryInterval` is set to 86400 (24 hours), in order to prevent exposing the broker to DOS attacks on hostile networks when using the out-of-the-box configuration (as an infinite expiry would allow an infinite number of retained/inflight messages to accumulate). If you are operating in a trusted environment, or you have capacity for a larger retention period, uou may wish to override this (set to `0` or `math.MaxInt` for no expiry).
+- By default, the value of `server.Options.Capabilities.MaximumMessageExpiryInterval` is set to 86400 (24 hours), in order to prevent exposing the broker to DOS attacks on hostile networks when using the out-of-the-box configuration (as an infinite expiry would allow an infinite number of retained/inflight messages to accumulate). If you are operating in a trusted environment, or you have capacity for a larger retention period, uou may wish to override this (set to `0` or `math.MaxInt64` for no expiry).
 
 ## Event Hooks 
 A universal event hooks system allows developers to hook into various parts of the server and client life cycle to add and modify functionality of the broker. These universal hooks are used to provide everything from authentication, persistent storage, to debugging tools.

clients.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -329,8 +330,14 @@ func (cl *Client) ResendInflightMessages(force bool) error {
 // ClearInflights deletes all inflight messages for the client, e.g. for a disconnected user with a clean session.
 func (cl *Client) ClearInflights(now, maximumExpiry int64) []uint16 {
 	deleted := []uint16{}
+
+	if maximumExpiry == 0 || maximumExpiry == math.MaxInt64 {
+		// If the maximum message expiry interval is set to 0 or math.MaxInt64, do not process expired inflight messages.
+		return deleted
+	}
+
 	for _, tk := range cl.State.Inflight.GetAll(false) {
-		if (tk.Expiry > 0 && tk.Expiry < now) || tk.Created+maximumExpiry < now {
+		if (tk.Expiry > 0 && tk.Expiry < now) || tk.Created < now-maximumExpiry {
 			if ok := cl.State.Inflight.Delete(tk.PacketID); ok {
 				cl.ops.hooks.OnQosDropped(cl, tk)
 				atomic.AddInt64(&cl.ops.info.Inflight, -1)

clients_test.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"io"
 	"log/slog"
+	"math"
 	"net"
 	"strings"
 	"sync/atomic"
@@ -315,6 +316,17 @@ func TestClientClearInflights(t *testing.T) {
 	require.Len(t, deleted, 3)
 	require.ElementsMatch(t, []uint16{1, 2, 5}, deleted)
 	require.Equal(t, 2, cl.State.Inflight.Len())
+
+	cl.State.Inflight.Set(packets.Packet{PacketID: 8, Created: n - 8})
+	deleted = cl.ClearInflights(n, 0)
+	require.Len(t, deleted, 0)
+	require.ElementsMatch(t, []uint16{}, deleted)
+	require.Equal(t, 3, cl.State.Inflight.Len())
+
+	deleted = cl.ClearInflights(n, math.MaxInt64)
+	require.Len(t, deleted, 0)
+	require.ElementsMatch(t, []uint16{}, deleted)
+	require.Equal(t, 3, cl.State.Inflight.Len())
 }
 
 func TestClientResendInflightMessages(t *testing.T) {

server.go

@@ -1596,8 +1596,15 @@ func (s *Server) clearExpiredClients(dt int64) {
 
 // clearExpiredRetainedMessage deletes retained messages from topics if they have expired.
 func (s *Server) clearExpiredRetainedMessages(now int64) {
+
+	if s.Options.Capabilities.MaximumMessageExpiryInterval == 0 ||
+		s.Options.Capabilities.MaximumMessageExpiryInterval == math.MaxInt64 {
+		// If the maximum message expiry interval is set to 0 or math.MaxInt64, do not process expired messages.
+		return
+	}
+
 	for filter, pk := range s.Topics.Retained.GetAll() {
-		if (pk.Expiry > 0 && pk.Expiry < now) || pk.Created+s.Options.Capabilities.MaximumMessageExpiryInterval < now {
+		if (pk.Expiry > 0 && pk.Expiry < now) || pk.Created < now-s.Options.Capabilities.MaximumMessageExpiryInterval {
 			s.Topics.Retained.Delete(filter)
 			s.hooks.OnRetainedExpired(filter)
 		}
@@ -1606,6 +1613,13 @@ func (s *Server) clearExpiredRetainedMessages(now int64) {
 
 // clearExpiredInflights deletes any inflight messages which have expired.
 func (s *Server) clearExpiredInflights(now int64) {
+
+	if s.Options.Capabilities.MaximumMessageExpiryInterval == 0 ||
+		s.Options.Capabilities.MaximumMessageExpiryInterval == math.MaxInt64 {
+		// If the maximum message expiry interval is set to 0 or math.MaxInt64, do not process expired messages.
+		return
+	}
+
 	for _, client := range s.Clients.GetAll() {
 		if deleted := client.ClearInflights(now, s.Options.Capabilities.MaximumMessageExpiryInterval); len(deleted) > 0 {
 			for _, id := range deleted {

server_test.go

@@ -9,6 +9,7 @@ import (
 	"encoding/binary"
 	"io"
 	"log/slog"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -3259,6 +3260,15 @@ func TestServerClearExpiredInflights(t *testing.T) {
 	s.clearExpiredInflights(n)
 	require.Len(t, cl.State.Inflight.GetAll(false), 2)
 	require.Equal(t, int64(-3), s.Info.Inflight)
+
+	s.Options.Capabilities.MaximumMessageExpiryInterval = math.MaxInt64
+	cl.State.Inflight.Set(packets.Packet{PacketID: 8, Expiry: n - 8})
+	s.clearExpiredInflights(n)
+	require.Len(t, cl.State.Inflight.GetAll(false), 3)
+
+	s.Options.Capabilities.MaximumMessageExpiryInterval = 0
+	s.clearExpiredInflights(n)
+	require.Len(t, cl.State.Inflight.GetAll(false), 3)
 }
 
 func TestServerClearExpiredRetained(t *testing.T) {
@@ -3276,6 +3286,15 @@ func TestServerClearExpiredRetained(t *testing.T) {
 	require.Len(t, s.Topics.Retained.GetAll(), 5)
 	s.clearExpiredRetainedMessages(n)
 	require.Len(t, s.Topics.Retained.GetAll(), 2)
+
+	s.Options.Capabilities.MaximumMessageExpiryInterval = math.MaxInt64
+	s.Topics.Retained.Add("o/p/q", packets.Packet{Created: n - 8})
+	s.clearExpiredRetainedMessages(n)
+	require.Len(t, s.Topics.Retained.GetAll(), 3)
+
+	s.Options.Capabilities.MaximumMessageExpiryInterval = 0
+	s.clearExpiredRetainedMessages(n)
+	require.Len(t, s.Topics.Retained.GetAll(), 3)
 }
 
 func TestServerClearExpiredClients(t *testing.T) {