adeelahmad · Oct 10, 2026 · Oct 10, 2026
diff --git a/fastapi_backend/app/domain/mqtt.py b/fastapi_backend/app/domain/mqtt.py
@@ -286,11 +286,14 @@ def _serve(self, sock, address):
                     topic, payload, qos, retain, pid = parse_publish(flags, body)
                     if not topic or "+" in topic or "#" in topic:
                         raise ProtocolError("a topic name can't have wildcards")
+                    # kept and passed on before the sender hears it arrived, so a client subscribing after the
+                    # acknowledgement gets it as the retained message
+                    self.publish(topic, payload, retain)
                     if qos == 1:
                         session.send(packet(PUBACK, 0, struct.pack("!H", pid)))
                     elif qos == 2:
                         session.send(packet(PUBREC, 0, struct.pack("!H", pid)))
-                    self._take(topic, payload, retain, session.login, address)
+                    self._hear(topic, payload, session.login, address)
                 elif kind == PUBREL:
                     session.send(packet(PUBCOMP, 0, body[:2]))
                 elif kind == SUBSCRIBE:
@@ -329,6 +332,9 @@ def _serve(self, sock, address):
 
     def _take(self, topic, payload, retain, login, address):
         self.publish(topic, payload, retain)
+        self._hear(topic, payload, login, address)
+
+    def _hear(self, topic, payload, login, address):
         if not topic.startswith("$"):
             try:
                 self.on_message(topic, payload, login, address)

diff --git a/fastapi_backend/tests/api/test_sensors.py b/fastapi_backend/tests/api/test_sensors.py
@@ -188,6 +188,25 @@ def test_broker_is_an_ordinary_local_broker():
         b.stop()
 
 
+def test_a_retained_message_is_kept_before_its_sender_hears_it_arrived():
+    b = mqtt.Broker("127.0.0.1", 0, lambda *a: {"ok": 1}, lambda *a: None).start()
+    keep = b.publish
+
+    def slow(*a, **kw):  # a busy hub: the acknowledgement must still wait for the message to be kept
+        time.sleep(0.3)
+        keep(*a, **kw)
+
+    b.publish = slow
+    try:
+        Device(b.port, "pub", "u", "p").publish("home/kitchen/temp", "21.5", qos=1, retain=True)
+        sub = Device(b.port, "sub", "u", "p")
+        assert sub.subscribe("home/#") == [0]
+        assert sub.receive() == ("home/kitchen/temp", b"21.5", True)
+        sub.close()
+    finally:
+        b.stop()
+
+
 def test_bridge_client_reads_another_broker():
     b = mqtt.Broker("127.0.0.1", 0, lambda *a: {"ok": 1}, lambda *a: None).start()
     try: