Allow configurable max stored qos > 0 messages (#359)

* Allow configurable max stored qos > 0 messages

* Only rollback Inflight if QoS > 0

* Only rollback Inflight if QoS > 0

* Minor refactor

Author: thedevop

clients.go

@@ -215,6 +215,10 @@ func (cl *Client) ParseConnect(lid string, pk packets.Packet) {
 	cl.Properties.Clean = pk.Connect.Clean
 	cl.Properties.Props = pk.Properties.Copy(false)
 
+	if cl.Properties.Props.ReceiveMaximum > cl.ops.options.Capabilities.MaximumInflight { // 3.3.4 Non-normative
+		cl.Properties.Props.ReceiveMaximum = cl.ops.options.Capabilities.MaximumInflight
+	}
+
 	if pk.Connect.Keepalive <= minimumKeepalive {
 		cl.ops.log.Warn(
 			ErrMinimumKeepalive.Error(),

clients_test.go

@@ -37,6 +37,7 @@ func newTestClient() (cl *Client, r net.Conn, w net.Conn) {
 		options: &Options{
 			Capabilities: &Capabilities{
 				ReceiveMaximum:             10,
+				MaximumInflight:            5,
 				TopicAliasMaximum:          10000,
 				MaximumClientWritesPending: 3,
 				maximumPacketID:            10,
@@ -183,6 +184,45 @@ func TestClientParseConnect(t *testing.T) {
 	require.Equal(t, int32(pk.Properties.ReceiveMaximum), cl.State.Inflight.maximumSendQuota)
 }
 
+func TestClientParseConnectReceiveMaxExceedMaxInflight(t *testing.T) {
+	const MaxInflight uint16 = 1
+	cl, _, _ := newTestClient()
+	cl.ops.options.Capabilities.MaximumInflight = MaxInflight
+
+	pk := packets.Packet{
+		ProtocolVersion: 4,
+		Connect: packets.ConnectParams{
+			ProtocolName:     []byte{'M', 'Q', 'T', 'T'},
+			Clean:            true,
+			Keepalive:        60,
+			ClientIdentifier: "mochi",
+			WillFlag:         true,
+			WillTopic:        "lwt",
+			WillPayload:      []byte("lol gg"),
+			WillQos:          1,
+			WillRetain:       false,
+		},
+		Properties: packets.Properties{
+			ReceiveMaximum: uint16(5),
+		},
+	}
+
+	cl.ParseConnect("tcp1", pk)
+	require.Equal(t, pk.Connect.ClientIdentifier, cl.ID)
+	require.Equal(t, pk.Connect.Keepalive, cl.State.Keepalive)
+	require.Equal(t, pk.Connect.Clean, cl.Properties.Clean)
+	require.Equal(t, pk.Connect.ClientIdentifier, cl.ID)
+	require.Equal(t, pk.Connect.WillTopic, cl.Properties.Will.TopicName)
+	require.Equal(t, pk.Connect.WillPayload, cl.Properties.Will.Payload)
+	require.Equal(t, pk.Connect.WillQos, cl.Properties.Will.Qos)
+	require.Equal(t, pk.Connect.WillRetain, cl.Properties.Will.Retain)
+	require.Equal(t, uint32(1), cl.Properties.Will.Flag)
+	require.Equal(t, int32(cl.ops.options.Capabilities.ReceiveMaximum), cl.State.Inflight.receiveQuota)
+	require.Equal(t, int32(cl.ops.options.Capabilities.ReceiveMaximum), cl.State.Inflight.maximumReceiveQuota)
+	require.Equal(t, int32(MaxInflight), cl.State.Inflight.sendQuota)
+	require.Equal(t, int32(MaxInflight), cl.State.Inflight.maximumSendQuota)
+}
+
 func TestClientParseConnectOverrideWillDelay(t *testing.T) {
 	cl, _, _ := newTestClient()
 

server.go

@@ -43,20 +43,21 @@ var (
 
 // Capabilities indicates the capabilities and features provided by the server.
 type Capabilities struct {
-	MaximumMessageExpiryInterval int64
-	MaximumClientWritesPending   int32
-	MaximumSessionExpiryInterval uint32
-	MaximumPacketSize            uint32
+	MaximumMessageExpiryInterval int64  // maximum message expiry if message expiry is 0 or over
+	MaximumClientWritesPending   int32  // maximum number of pending message writes for a client
+	MaximumSessionExpiryInterval uint32 // maximum number of seconds to keep disconnected sessions
+	MaximumPacketSize            uint32 // maximum packet size, no limit if 0
 	maximumPacketID              uint32 // unexported, used for testing only
-	ReceiveMaximum               uint16
-	TopicAliasMaximum            uint16
-	SharedSubAvailable           byte
-	MinimumProtocolVersion       byte
+	ReceiveMaximum               uint16 // maximum number of concurrent qos messages per client
+	MaximumInflight              uint16 // maximum number of qos > 0 messages can be stored, 0(=8192)-65535
+	TopicAliasMaximum            uint16 // maximum topic alias value
+	SharedSubAvailable           byte   // support of shared subscriptions
+	MinimumProtocolVersion       byte   // minimum supported mqtt version
 	Compatibilities              Compatibilities
-	MaximumQos                   byte
-	RetainAvailable              byte
-	WildcardSubAvailable         byte
-	SubIDAvailable               byte
+	MaximumQos                   byte // maximum qos value available to clients
+	RetainAvailable              byte // support of retain messages
+	WildcardSubAvailable         byte // support of wildcard subscriptions
+	SubIDAvailable               byte // support of subscription identifiers
 }
 
 // NewDefaultServerCapabilities defines the default features and capabilities provided by the server.
@@ -68,6 +69,7 @@ func NewDefaultServerCapabilities() *Capabilities {
 		MaximumPacketSize:            0,              // no maximum packet size
 		maximumPacketID:              math.MaxUint16,
 		ReceiveMaximum:               1024,           // maximum number of concurrent qos messages per client
+    MaximumInflight:              1024 * 8,       // maximum number of qos > 0 messages can be stored
 		TopicAliasMaximum:            math.MaxUint16, // maximum topic alias value
 		SharedSubAvailable:           1,              // shared subscriptions are available
 		MinimumProtocolVersion:       3,              // minimum supported mqtt version (3.0.0)
@@ -201,6 +203,10 @@ func (o *Options) ensureDefaults() {
 
 	o.Capabilities.maximumPacketID = math.MaxUint16 // spec maximum is 65535
 
+	if o.Capabilities.MaximumInflight == 0 {
+		o.Capabilities.MaximumInflight = 1024 * 8
+	}
+
 	if o.SysTopicResendInterval == 0 {
 		o.SysTopicResendInterval = defaultSysTopicInterval
 	}
@@ -975,9 +981,17 @@ func (s *Server) publishToClient(cl *Client, sub packets.Subscription, pk packet
 	}
 
 	if out.FixedHeader.Qos > 0 {
+		if cl.State.Inflight.Len() >= int(s.Options.Capabilities.MaximumInflight) {
+			// add hook?
+			atomic.AddInt64(&s.Info.InflightDropped, 1)
+			s.Log.Warn("client store quota reached", "client", cl.ID, "listener", cl.Net.Listener)
+			return out, packets.ErrQuotaExceeded
+		}
+
 		i, err := cl.NextPacketID() // [MQTT-4.3.2-1] [MQTT-4.3.3-1]
 		if err != nil {
 			s.hooks.OnPacketIDExhausted(cl, pk)
+			atomic.AddInt64(&s.Info.InflightDropped, 1)
 			s.Log.Warn("packet ids exhausted", "error", err, "client", cl.ID, "listener", cl.Net.Listener)
 			return out, packets.ErrQuotaExceeded
 		}
@@ -1008,8 +1022,10 @@ func (s *Server) publishToClient(cl *Client, sub packets.Subscription, pk packet
 	default:
 		atomic.AddInt64(&s.Info.MessagesDropped, 1)
 		cl.ops.hooks.OnPublishDropped(cl, pk)
-		cl.State.Inflight.Delete(out.PacketID) // packet was dropped due to irregular circumstances, so rollback inflight.
-		cl.State.Inflight.IncreaseSendQuota()
+		if out.FixedHeader.Qos > 0 {
+			cl.State.Inflight.Delete(out.PacketID) // packet was dropped due to irregular circumstances, so rollback inflight.
+			cl.State.Inflight.IncreaseSendQuota()
+		}
 		return out, packets.ErrPendingClientWritesExceeded
 	}
 

server_test.go

@@ -1907,6 +1907,7 @@ func TestPublishToClientSubscriptionDowngradeQos(t *testing.T) {
 }
 
 func TestPublishToClientExceedClientWritesPending(t *testing.T) {
+	var sendQuota uint16 = 5
 	s := newServer()
 
 	_, w := net.Pipe()
@@ -1917,9 +1918,12 @@ func TestPublishToClientExceedClientWritesPending(t *testing.T) {
 		options: &Options{
 			Capabilities: &Capabilities{
 				MaximumClientWritesPending: 3,
+				maximumPacketID:            10,
 			},
 		},
 	})
+	cl.Properties.Props.ReceiveMaximum = sendQuota
+	cl.State.Inflight.ResetSendQuota(int32(cl.Properties.Props.ReceiveMaximum))
 
 	s.Clients.Add(cl)
 
@@ -1928,9 +1932,20 @@ func TestPublishToClientExceedClientWritesPending(t *testing.T) {
 		atomic.AddInt32(&cl.State.outboundQty, 1)
 	}
 
+	id, _ := cl.NextPacketID()
+	cl.State.Inflight.Set(packets.Packet{PacketID: uint16(id)})
+	cl.State.Inflight.DecreaseSendQuota()
+	sendQuota--
+
 	_, err := s.publishToClient(cl, packets.Subscription{Filter: "a/b/c", Qos: 2}, packets.Packet{})
 	require.Error(t, err)
 	require.ErrorIs(t, packets.ErrPendingClientWritesExceeded, err)
+	require.Equal(t, int32(sendQuota), atomic.LoadInt32(&cl.State.Inflight.sendQuota))
+
+	_, err = s.publishToClient(cl, packets.Subscription{Filter: "a/b/c", Qos: 2}, packets.Packet{FixedHeader: packets.FixedHeader{Qos: 1}})
+	require.Error(t, err)
+	require.ErrorIs(t, packets.ErrPendingClientWritesExceeded, err)
+	require.Equal(t, int32(sendQuota), atomic.LoadInt32(&cl.State.Inflight.sendQuota))
 }
 
 func TestPublishToClientServerTopicAlias(t *testing.T) {
@@ -1986,6 +2001,22 @@ func TestPublishToClientMqtt5RetainAsPublishedTrueLeverageNoConn(t *testing.T) {
 	require.ErrorIs(t, err, packets.CodeDisconnect)
 }
 
+func TestPublishToClientExceedMaximumInflight(t *testing.T) {
+	const MaxInflight uint16 = 5
+	s := newServer()
+	cl, _, _ := newTestClient()
+	s.Options.Capabilities.MaximumInflight = MaxInflight
+	cl.ops.options.Capabilities.MaximumInflight = MaxInflight
+	for i := uint16(0); i < MaxInflight; i++ {
+		cl.State.Inflight.Set(packets.Packet{PacketID: i})
+	}
+
+	_, err := s.publishToClient(cl, packets.Subscription{Filter: "a/b/c", Qos: 1}, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
+	require.Error(t, err)
+	require.ErrorIs(t, err, packets.ErrQuotaExceeded)
+	require.Equal(t, int64(1), atomic.LoadInt64(&s.Info.InflightDropped))
+}
+
 func TestPublishToClientExhaustedPacketID(t *testing.T) {
 	s := newServer()
 	cl, _, _ := newTestClient()
@@ -1996,6 +2027,7 @@ func TestPublishToClientExhaustedPacketID(t *testing.T) {
 	_, err := s.publishToClient(cl, packets.Subscription{Filter: "a/b/c", Qos: 1}, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
 	require.Error(t, err)
 	require.ErrorIs(t, err, packets.ErrQuotaExceeded)
+	require.Equal(t, int64(1), atomic.LoadInt64(&s.Info.InflightDropped))
 }
 
 func TestPublishToClientACLNotAuthorized(t *testing.T) {