Differentiate the handling of 'expire' in MQTTv5 and MQTTv3; skip exp…

…iration checks if MaximumMessageExpiryInterval is set to 0; optimize code and test cases.

README-CN.md

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

@@ -329,8 +329,12 @@ 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{}
+
 	for _, tk := range cl.State.Inflight.GetAll(false) {
-		if (tk.Expiry > 0 && tk.Expiry < now) || now-tk.Created > maximumExpiry {
+		expired := tk.ProtocolVersion == 5 && (tk.Expiry > 0 && tk.Expiry < now) // [MQTT-3.3.2-5]
+		abandoned := now-tk.Created > maximumExpiry
+
+		if expired || abandoned {
 			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

@@ -304,17 +304,28 @@ func TestClientClearInflights(t *testing.T) {
 	cl, _, _ := newTestClient()
 
 	n := time.Now().Unix()
-	cl.State.Inflight.Set(packets.Packet{PacketID: 1, Expiry: n - 1})
-	cl.State.Inflight.Set(packets.Packet{PacketID: 2, Expiry: n - 2})
-	cl.State.Inflight.Set(packets.Packet{PacketID: 3, Created: n - 3}) // within bounds
-	cl.State.Inflight.Set(packets.Packet{PacketID: 5, Created: n - 5}) // over max server expiry limit
-	cl.State.Inflight.Set(packets.Packet{PacketID: 7, Created: n})
+	cl.State.Inflight.Set(packets.Packet{ProtocolVersion: 5, PacketID: 1, Expiry: n - 1})
+	cl.State.Inflight.Set(packets.Packet{ProtocolVersion: 5, PacketID: 2, Expiry: n - 2})
+	cl.State.Inflight.Set(packets.Packet{ProtocolVersion: 5, PacketID: 3, Created: n - 3}) // within bounds
+	cl.State.Inflight.Set(packets.Packet{ProtocolVersion: 5, PacketID: 5, Created: n - 5}) // over max server expiry limit
+	cl.State.Inflight.Set(packets.Packet{ProtocolVersion: 5, PacketID: 7, Created: n})
 	require.Equal(t, 5, cl.State.Inflight.Len())
 
 	deleted := cl.ClearInflights(n, 4)
 	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: 11, Expiry: n - 1})
+	cl.State.Inflight.Set(packets.Packet{PacketID: 12, Expiry: n - 2})  // expiry is ineffective for v3.
+	cl.State.Inflight.Set(packets.Packet{PacketID: 13, Created: n - 3}) // within bounds for v3
+	cl.State.Inflight.Set(packets.Packet{PacketID: 15, Created: n - 5}) // over max server expiry limit
+	require.Equal(t, 6, cl.State.Inflight.Len())
+
+	deleted = cl.ClearInflights(n, 4)
+	require.Len(t, deleted, 3)
+	require.ElementsMatch(t, []uint16{11, 12, 15}, deleted)
+	require.Equal(t, 3, cl.State.Inflight.Len())
 }
 
 func TestClientResendInflightMessages(t *testing.T) {

server.go

@@ -1596,13 +1596,17 @@ func (s *Server) clearExpiredClients(dt int64) {
 
 // clearExpiredRetainedMessage deletes retained messages from topics if they have expired.
 func (s *Server) clearExpiredRetainedMessages(now int64) {
-	// If the maximum message expiry interval is set to math.MaxInt64, do not process expired messages.
-	if s.Options.Capabilities.MaximumMessageExpiryInterval == math.MaxInt64 {
+
+	// If the maximum message expiry interval is set to 0, do not process expired messages for all protocols.
+	if s.Options.Capabilities.MaximumMessageExpiryInterval == 0 {
 		return
 	}
 
 	for filter, pk := range s.Topics.Retained.GetAll() {
-		if (pk.Expiry > 0 && pk.Expiry < now) || now-pk.Created > s.Options.Capabilities.MaximumMessageExpiryInterval {
+		expired := pk.ProtocolVersion == 5 && (pk.Expiry > 0 && pk.Expiry < now) // [MQTT-3.3.2-5]
+		abandoned := now-pk.Created > s.Options.Capabilities.MaximumMessageExpiryInterval
+
+		if expired || abandoned {
 			s.Topics.Retained.Delete(filter)
 			s.hooks.OnRetainedExpired(filter)
 		}
@@ -1611,8 +1615,9 @@ func (s *Server) clearExpiredRetainedMessages(now int64) {
 
 // clearExpiredInflights deletes any inflight messages which have expired.
 func (s *Server) clearExpiredInflights(now int64) {
-	// If the maximum message expiry interval is set math.MaxInt64, do not process expired messages.
-	if s.Options.Capabilities.MaximumMessageExpiryInterval == math.MaxInt64 {
+
+	// If the maximum message expiry interval is set to 0, do not process expired messages for all protocols.
+	if s.Options.Capabilities.MaximumMessageExpiryInterval == 0 {
 		return
 	}
 

server_test.go

@@ -3273,20 +3273,28 @@ func TestServerClearExpiredRetained(t *testing.T) {
 	s.Options.Capabilities.MaximumMessageExpiryInterval = 4
 
 	n := time.Now().Unix()
-	s.Topics.Retained.Add("a/b/c", packets.Packet{Created: n, Expiry: n - 1})
-	s.Topics.Retained.Add("d/e/f", packets.Packet{Created: n, Expiry: n - 2})
-	s.Topics.Retained.Add("g/h/i", packets.Packet{Created: n - 3}) // within bounds
-	s.Topics.Retained.Add("j/k/l", packets.Packet{Created: n - 5}) // over max server expiry limit
-	s.Topics.Retained.Add("m/n/o", packets.Packet{Created: n})
+	s.Topics.Retained.Add("a/b/c", packets.Packet{ProtocolVersion: 5, Created: n, Expiry: n - 1})
+	s.Topics.Retained.Add("d/e/f", packets.Packet{ProtocolVersion: 5, Created: n, Expiry: n - 2})
+	s.Topics.Retained.Add("g/h/i", packets.Packet{ProtocolVersion: 5, Created: n - 3}) // within bounds
+	s.Topics.Retained.Add("j/k/l", packets.Packet{ProtocolVersion: 5, Created: n - 5}) // over max server expiry limit
+	s.Topics.Retained.Add("m/n/o", packets.Packet{ProtocolVersion: 5, Created: n})
 
 	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.Topics.Retained.Add("p/q/r", packets.Packet{Created: n, Expiry: n - 1})
+	s.Topics.Retained.Add("s/t/u", packets.Packet{Created: n, Expiry: n - 2}) // expiry is ineffective for v3.
+	s.Topics.Retained.Add("v/w/x", packets.Packet{Created: n - 3})            // within bounds for v3
+	s.Topics.Retained.Add("y/z/1", packets.Packet{Created: n - 5})            // over max server expiry limit
+	require.Len(t, s.Topics.Retained.GetAll(), 6)
+	s.clearExpiredRetainedMessages(n)
+	require.Len(t, s.Topics.Retained.GetAll(), 5)
+
+	s.Options.Capabilities.MaximumMessageExpiryInterval = 0
+	s.Topics.Retained.Add("2/3/4", packets.Packet{Created: n - 8})
 	s.clearExpiredRetainedMessages(n)
-	require.Len(t, s.Topics.Retained.GetAll(), 3)
+	require.Len(t, s.Topics.Retained.GetAll(), 6)
 }
 
 func TestServerClearExpiredClients(t *testing.T) {