Fix QoS 2 message rejections to send PUBREC instead of PUBACK, allow errRejectPacket through

server.go

@@ -913,16 +913,30 @@ func (s *Server) processPublish(cl *Client, pk packets.Packet) error {
 	pkx, err := s.hooks.OnPublish(cl, pk)
 	if err == nil {
 		pk = pkx
-	} else if errors.Is(err, packets.ErrRejectPacket) {
-		return nil
 	} else if errors.Is(err, packets.CodeSuccessIgnore) {
 		pk.Ignore = true
-	} else if cl.Properties.ProtocolVersion == 5 && pk.FixedHeader.Qos > 0 && errors.As(err, new(packets.Code)) {
-		err = cl.WritePacket(s.buildAck(pk.PacketID, packets.Puback, 0, pk.Properties, err.(packets.Code)))
-		if err != nil {
-			return err
+	} else if errors.As(err, new(packets.Code)) {
+		// No ack, do not publish to subscribers. Swallow the error.
+		if pk.FixedHeader.Qos == 0 {
+			return nil
+
+			// qos1&2, client version 5. Send puback or pubrec. Do not publish to subscribers.
+		} else if cl.Properties.ProtocolVersion == 5 {
+			ackType := packets.Puback
+			if pk.FixedHeader.Qos == 2 {
+				ackType = packets.Pubrec
+			}
+			err = cl.WritePacket(s.buildAck(pk.PacketID, ackType, 0, pk.Properties, err.(packets.Code)))
+			if err != nil {
+				return err
+			}
+			return nil
 		}
-		return nil
+		// qos 1&2, client version < 5
+		return s.DisconnectClient(cl, err.(packets.Code)) //  [MQTT-3.3.5-2] https://github.com/mochi-mqtt/server/issues/290#issuecomment-1709193269
+	} else {
+		// error was not a packets.Code, so we return it as is.
+		return err
 	}
 
 	if pk.FixedHeader.Retain { // [MQTT-3.3.1-5] ![MQTT-3.3.1-8]

server_test.go

@@ -1739,7 +1739,8 @@ func TestServerProcessPublishACLCheckDeny(t *testing.T) {
 	}
 }
 
-func TestServerProcessPublishOnMessageRecvRejected(t *testing.T) {
+func TestServerProcessPublishOnMessageRecvRejectedQos0(t *testing.T) {
+	// Test that QoS 0 rejections are silently ignored (return nil)
 	s := newServer()
 	require.NotNil(t, s)
 	hook := new(modifiedHookBase)
@@ -1752,8 +1753,86 @@ func TestServerProcessPublishOnMessageRecvRejected(t *testing.T) {
 	_ = s.Serve()
 	defer s.Close()
 	cl, _, _ := newTestClient()
+
+	// QoS 0 rejections should be silently ignored (return nil)
 	err = s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet)
-	require.NoError(t, err) // packets rejected silently
+	require.NoError(t, err)
+}
+
+func TestServerProcessPublishOnMessageRecvRejectedQos1(t *testing.T) {
+	s := newServer()
+	require.NotNil(t, s)
+	hook := new(modifiedHookBase)
+	hook.fail = true
+	hook.err = packets.ErrRejectPacket
+
+	err := s.AddHook(hook, nil)
+	require.NoError(t, err)
+
+	_ = s.Serve()
+	defer s.Close()
+	cl, r, w := newTestClient()
+	cl.Properties.ProtocolVersion = 5
+
+	go func() {
+		err := s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
+		require.NoError(t, err)
+		_ = w.Close()
+	}()
+
+	buf, err := io.ReadAll(r)
+	require.NoError(t, err)
+
+	// Should receive PUBACK with reject reason code 0x83 and reason string
+	expected := []byte{
+		byte(packets.Puback << 4), // packet type
+		22,                        // remaining length
+		0, 7,                      // packet id
+		packets.ErrRejectPacket.Code, // reason code 0x83
+		18,                           // properties length
+		31,                           // reason string property ID
+		0, 15,                        // reason string length
+		'p', 'a', 'c', 'k', 'e', 't', ' ', 'r', 'e', 'j', 'e', 'c', 't', 'e', 'd', // "packet rejected"
+	}
+	require.Equal(t, expected, buf)
+}
+
+func TestServerProcessPublishOnMessageRecvRejectedQos2(t *testing.T) {
+	s := newServer()
+	require.NotNil(t, s)
+	hook := new(modifiedHookBase)
+	hook.fail = true
+	hook.err = packets.ErrRejectPacket
+
+	err := s.AddHook(hook, nil)
+	require.NoError(t, err)
+
+	_ = s.Serve()
+	defer s.Close()
+	cl, r, w := newTestClient()
+	cl.Properties.ProtocolVersion = 5
+
+	go func() {
+		err := s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos2).Packet)
+		require.NoError(t, err)
+		_ = w.Close()
+	}()
+
+	buf, err := io.ReadAll(r)
+	require.NoError(t, err)
+
+	// Should receive PUBREC with reject reason code 0x83 and reason string
+	expected := []byte{
+		byte(packets.Pubrec << 4), // packet type
+		22,                        // remaining length
+		0, 7,                      // packet id
+		packets.ErrRejectPacket.Code, // reason code 0x83
+		18,                           // properties length
+		31,                           // reason string property ID
+		0, 15,                        // reason string length
+		'p', 'a', 'c', 'k', 'e', 't', ' ', 'r', 'e', 'j', 'e', 'c', 't', 'e', 'd', // "packet rejected"
+	}
+	require.Equal(t, expected, buf)
 }
 
 func TestServerProcessPacketPublishQos0(t *testing.T) {