diff --git a/Changelog.md b/Changelog.md
--- a/Changelog.md
+++ b/Changelog.md
@@ -1,5 +1,16 @@
 # Changelog for net-mqtt
 
+## 0.2.4.1
+
+Link QoS2 completion thread on subscriber.
+
+An exception from a subscriber callback could be silently dropped
+without completing the handshake which would then cause the MQTT
+broker to just stop sending messages to the subscriber.
+Unfortunately, the broker (at least mosquitto) still responds to pings
+and doesn't give any useful notification that it's no longer sending
+messages.
+
 ## 0.2.4.0
 
 Introduced `Filter` type alias to distinguish from `Topic`.
diff --git a/net-mqtt.cabal b/net-mqtt.cabal
--- a/net-mqtt.cabal
+++ b/net-mqtt.cabal
@@ -4,10 +4,10 @@
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 823f507cc62f2e9bd1f915a491b6404b412b35e9c9848e46e1927c1533ed6b0f
+-- hash: dc7e4a1db1c001ec1c63bdbb3da55fd4ff4a6f42aaf09aac10288f50eb11d858
 
 name:           net-mqtt
-version:        0.2.4.0
+version:        0.2.4.1
 synopsis:       An MQTT Protocol Implementation.
 description:    Please see the README on GitHub at <https://github.com/dustin/mqtt-hs#readme>
 category:       Network
diff --git a/src/Network/MQTT/Client.hs b/src/Network/MQTT/Client.hs
--- a/src/Network/MQTT/Client.hs
+++ b/src/Network/MQTT/Client.hs
@@ -255,7 +255,7 @@
             Just ch -> writeTChan ch pkt
 
         pubMachine PublishRequest{..}
-          | _pubQoS == QoS2 = void $ async manageQoS2
+          | _pubQoS == QoS2 = void $ async manageQoS2 >>= link
           | _pubQoS == QoS1 = notify >> sendPacketIO c (PubACKPkt (PubACK _pubPktID))
           | otherwise = notify
 
@@ -269,12 +269,20 @@
               atomically $ modifyTVar' _acks (Map.insert (DPubREL, _pubPktID) ch)
               E.finally (manageQoS2' ch) (atomically $ releasePktID c (DPubREL, _pubPktID))
                 where
-                  manageQoS2' ch = do
+                  sendREC ch = do
                     sendPacketIO c (PubRECPkt (PubREC _pubPktID))
                     (PubRELPkt _) <- atomically $ readTChan ch
-                    notify
-                    sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID))
+                    pure ()
 
+                  manageQoS2' ch = do
+                    v <- timeout 10000000 (sendREC ch)
+                    case v of
+                      Nothing -> killConn c Timeout
+                      _ ->  notify >> sendPacketIO c (PubCOMPPkt (PubCOMP _pubPktID))
+
+killConn :: E.Exception e => MQTTClient -> e -> IO ()
+killConn MQTTClient{..} e = readTVarIO _ct >>= \t -> cancelWith t e
+
 sendPacket :: MQTTClient -> MQTTPkt -> STM ()
 sendPacket MQTTClient{..} p = do
   st <- readTVar _st
@@ -385,7 +393,7 @@
 disconnect c@MQTTClient{..} = race_ getDisconnected orDieTrying
   where
     getDisconnected = sendPacketIO c DisconnectPkt >> waitForClient c
-    orDieTrying = threadDelay 10000000 >> readTVarIO _ct >>= \t -> cancelWith t Timeout
+    orDieTrying = threadDelay 10000000 >> killConn c Timeout
 
 -- | A convenience method for creating a LastWill.
 mkLWT :: Topic -> BL.ByteString -> Bool -> T.LastWill