mqtt-notify: Recover from lost connectionsHEADmaster

The service runs for as long as the session does, so it has to survive
laptop suspends and network changes. It did not: a connection that died
while the machine was asleep could leave the client wedged, with no way
back short of restarting the unit by hand.

Tighten the reconnect behavior and add a way out when it still fails:

  - Bound paho's reconnect backoff to 30 seconds. The default grows to
    two minutes between attempts, so even a successful reconnect could
    take minutes to happen.

  - Drop the keepalive from 60 to 30 seconds so that a connection which
    went away without a FIN is noticed sooner.

  - Track how long we have been disconnected and exit if that exceeds
    two minutes, and set Restart=always in the unit file so that systemd
    brings us back with a fresh client. CLOCK_MONOTONIC excludes time
    spent suspended, so this does not fire spuriously on resume.

Also stop calling sys.exit() from on_message(). It runs on paho's
network thread, where SystemExit only kills that thread and leaves the
main loop running with no connection. Quit the main loop instead.

1 files changed, 59 insertions, 6 deletions

diff --git a/mqtt-notify.py b/mqtt-notify.py
index 287427f..27a2bb2 100755
--- a/mqtt-notify.py
+++ b/mqtt-notify.py
@@ -14,6 +14,7 @@ import configparser
 import re
 import signal
 import sys
+import threading
 import time
 import paho.mqtt.client as mqtt
 import gi
@@ -32,6 +33,15 @@ subj_fmt = re.compile(r"IRC message (on|from) (?P<key>.*)")
 
 notification_map = {}
 
+DISCONNECT_TIMEOUT = 120
+WATCHDOG_INTERVAL = 30
+KEEPALIVE = 30
+
+# Time (CLOCK_MONOTONIC, so time spent suspended does not count) since which
+# the client has been disconnected, or None while connected.
+state_lock = threading.Lock()
+disconnected_since = time.monotonic()
+
 
 class Signaler:
     def __init__(self, loop):
@@ -42,11 +52,19 @@ class Signaler:
 
 
 def on_connect(client, userdata, flags, reason_code, properties):
-    print("Connected")
+    global disconnected_since
+
+    print("Connected: {}".format(reason_code))
+
+    if reason_code.is_failure:
+        return
+
+    with state_lock:
+        disconnected_since = None
 
     # Subscribing in on_connect() means that if we lose the connection and
     # reconnect then subscriptions will be renewed.
-    client.subscribe(userdata)
+    client.subscribe(userdata["topic"])
 
 
 def on_close(notification):
@@ -95,11 +113,35 @@ def on_message(client, userdata, msg):
         n.show()
     except GLib.GError as e:
         print("Failed to show notification: {}".format(e), file=sys.stderr)
-        sys.exit(-1)
+        GLib.idle_add(userdata["loop"].quit)
 
 
 def on_disconnect(client, userdata, flags, reason_code, properties):
-    print("Disconnected")
+    global disconnected_since
+
+    print("Disconnected: {}".format(reason_code))
+
+    with state_lock:
+        if disconnected_since is None:
+            disconnected_since = time.monotonic()
+
+
+def watchdog(loop):
+    # paho reconnects on its own, but a connection that died while the machine
+    # was suspended or on another network can leave it wedged. If it has not
+    # come back after DISCONNECT_TIMEOUT, exit and let systemd restart us.
+    with state_lock:
+        since = disconnected_since
+
+    if since is not None and time.monotonic() - since > DISCONNECT_TIMEOUT:
+        print(
+            "Disconnected for more than {} seconds, exiting".format(DISCONNECT_TIMEOUT),
+            file=sys.stderr,
+        )
+        loop.quit()
+        return GLib.SOURCE_REMOVE
+
+    return GLib.SOURCE_CONTINUE
 
 
 def password(user, host):
@@ -164,17 +206,28 @@ def main(argv):
     user, broker, port, topic = config(args.config.name)
 
     Notify.init("MQTT to Notify bridge")
-    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, userdata=topic)
+    client = mqtt.Client(
+        mqtt.CallbackAPIVersion.VERSION2, userdata={"topic": topic, "loop": loop}
+    )
 
     client.tls_set()
     client.username_pw_set(user, password(user, broker))
     client.on_connect = on_connect
     client.on_message = on_message
     client.on_disconnect = on_disconnect
-    client.connect_async(broker, port, 60)
+
+    # Retry forever with a bounded backoff rather than paho's default, which
+    # grows to two minutes between attempts.
+    client.reconnect_delay_set(min_delay=1, max_delay=30)
+
+    # connect_async() plus loop_start() means the initial connection is also
+    # retried, so we come up even if the network is not ready yet.
+    client.connect_async(broker, port, KEEPALIVE)
 
     client.loop_start()
 
+    GLib.timeout_add_seconds(WATCHDOG_INTERVAL, watchdog, loop)
+
     loop.run()
 
     client.loop_stop()