Skip to content

Commit cf64e56

Browse files
haoruizhouhaoruizhou
committed
Measure pubsub liveness via a heartbeat channel and self-heal dead subscriptions
Publish a heartbeat on the telemetry_heartbeat pubsub channel every second and have each subscriber measure liveness on the pubsub connection itself: if no message arrives for HEARTBEAT_STALE_S, tear down and re-subscribe. An out-of-band key check can't see a half-dead pubsub connection because regular commands use a different pool connection that redis-py silently reconnects. Wrap the uplink relay in a reconnect loop so the pump returning actually re-subscribes instead of exiting, and pass shutdown_event.is_set as should_stop so SIGTERM stops the bridge cleanly. Add the heartbeat tests to CI.
1 parent 3b2b48d commit cf64e56

6 files changed

Lines changed: 215 additions & 183 deletions

File tree

‎.github/workflows/telemetry-ci.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,7 @@ jobs:
166166
tests/test_page_lock.py \
167167
tests/test_stats_publisher.py \
168168
tests/test_status_server.py \
169+
tests/test_heartbeat.py \
169170
-v
170171
171172
# ── Integration tests: full docker-compose stack ─────────────────────────

‎universal-telemetry-software/src/config.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@
2222
REDIS_STATS_CHANNEL = "system_stats"
2323
REDIS_DIAG_CHANNEL = "link_diagnostics"
2424
REDIS_WS_CLIENTS_KEY = "websocket_bridge:clients"
25-
REDIS_HEARTBEAT_KEY = "telemetry:heartbeat"
26-
HEARTBEAT_STALE_S = 5.0 # subscribers reconnect if heartbeat older than this
25+
REDIS_HEARTBEAT_CHANNEL = "telemetry_heartbeat"
26+
HEARTBEAT_STALE_S = 5.0 # subscribers reconnect if no pubsub message arrives for this long
2727

2828
# ── Feature flags ─────────────────────────────────────────────────────────────
2929
ENABLE_UPLINK = os.getenv("ENABLE_UPLINK", "false").lower() == "true"

‎universal-telemetry-software/src/data.py‎

Lines changed: 55 additions & 108 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import asyncio
2+
import contextlib
23
import socket
34
import struct
45
import time
@@ -14,10 +15,10 @@
1415
from src.config import (
1516
REMOTE_IP, UDP_PORT, TCP_PORT,
1617
REDIS_URL, REDIS_CAN_CHANNEL, REDIS_UPLINK_CHANNEL, ENABLE_UPLINK,
17-
REDIS_WS_CLIENTS_KEY,
18+
REDIS_WS_CLIENTS_KEY, REDIS_HEARTBEAT_CHANNEL,
1819
)
1920
from src import redis_utils, utils
20-
from src.heartbeat import run_heartbeat_writer
21+
from src.heartbeat import pump_pubsub_with_heartbeat, run_heartbeat_writer
2122
from src.version import get_git_hash
2223

2324
BATCH_SIZE = 20
@@ -90,66 +91,6 @@ def publish(self, channel, data):
9091
except asyncio.QueueFull:
9192
pass # drop under backpressure rather than block the CAN reader
9293

93-
@staticmethod
94-
async def _pump_pubsub_with_heartbeat(pubsub, redis_client, on_message, *,
95-
stale_s: float = 5.0,
96-
log=None):
97-
"""Drain a Redis pubsub, restarting it if the heartbeat goes stale.
98-
99-
Replaces the naive `async for message in pubsub.listen():` pattern that
100-
silently goes dark when the TCP connection or subscription state is lost
101-
after a car power-cycle. Uses non-blocking get_message(timeout=1.0) so we
102-
get a chance to check the heartbeat key on every iteration.
103-
104-
`on_message` is an `async` callable invoked with the raw pubsub message
105-
dict (whatever `get_message()` returns, typically
106-
`{"type": "message", "channel": ..., "data": ...}`) for each non-None,
107-
non-subscribe-confirmation message. Returns only on cancellation; outer
108-
supervisor is expected to restart the task.
109-
"""
110-
from .config import REDIS_HEARTBEAT_KEY
111-
_log = log or logger
112-
last_hb_mono = 0.0
113-
while True:
114-
try:
115-
msg = await pubsub.get_message(timeout=1.0, ignore_subscribe_messages=True)
116-
except asyncio.CancelledError:
117-
raise
118-
except Exception as e:
119-
_log.warning(f"pubsub.get_message error, reconnecting: {e}")
120-
await asyncio.sleep(0.5)
121-
return # let the outer while-True re-subscribe
122-
123-
if msg is not None:
124-
try:
125-
await on_message(msg)
126-
except Exception as e:
127-
_log.error(f"subscriber handler error: {e}")
128-
last_hb_mono = time.monotonic() # data flowing — heartbeat check not needed
129-
continue
130-
131-
# No message this second — check heartbeat
132-
try:
133-
raw = await redis_client.get(REDIS_HEARTBEAT_KEY) if redis_client else None
134-
except Exception as e:
135-
_log.debug(f"heartbeat read failed: {e}")
136-
raw = None
137-
138-
if raw is None:
139-
# First-time setup or Redis is down — don't reconnect, just wait.
140-
if last_hb_mono == 0.0:
141-
continue
142-
# Heartbeat missing while we've been running: producer is gone or
143-
# the key expired. Tear down so the outer loop re-subscribes.
144-
if time.monotonic() - last_hb_mono > stale_s:
145-
_log.warning("heartbeat stale, forcing pubsub reconnect")
146-
return
147-
else:
148-
last_hb_mono = time.monotonic()
149-
if time.monotonic() - last_hb_mono > stale_s:
150-
_log.warning("heartbeat stale, forcing pubsub reconnect")
151-
return
152-
15394
def _handle_ecu_timestamp(self, data: bytes) -> None:
15495
"""Extract epoch_ms from ECU VCU_Timestamp message and update clock offset."""
15596
if len(data) < 8:
@@ -609,8 +550,6 @@ async def uplink_receiver():
609550
throughput_listener_task(),
610551
utils.heartbeat_coro(self.telemetry_event),
611552
inject_heartbeat(),
612-
# Car has no Redis — writer is a clean no-op when redis_client is None.
613-
run_heartbeat_writer(None),
614553
]
615554
if ENABLE_UPLINK:
616555
tasks.append(uplink_receiver())
@@ -926,55 +865,63 @@ async def uplink_relay():
926865
uplink_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
927866
uplink_seq = 0
928867

929-
try:
930-
r = aioredis.from_url(REDIS_URL)
931-
pubsub = r.pubsub()
932-
await pubsub.subscribe(REDIS_UPLINK_CHANNEL)
933-
logger.info(f"Subscribed to Redis channel: {REDIS_UPLINK_CHANNEL}")
934-
935-
async def _relay(msg):
936-
nonlocal uplink_seq
937-
if msg['type'] != 'message':
938-
return
939-
try:
940-
data = redis_utils.decode_message(msg['data'])
941-
uplink_msg = json.loads(data)
942-
943-
can_id = uplink_msg.get("canId")
944-
can_data = uplink_msg.get("data", [])
945-
ref = uplink_msg.get("ref", "unknown")
868+
async def _relay(msg):
869+
nonlocal uplink_seq
870+
if msg['type'] != 'message':
871+
return
872+
try:
873+
data = redis_utils.decode_message(msg['data'])
874+
uplink_msg = json.loads(data)
946875

947-
if can_id is None or not isinstance(can_id, int) or can_id < 0:
948-
logger.warning(f"Uplink relay: invalid canId in ref={ref}")
949-
return
950-
if not isinstance(can_data, list) or len(can_data) < 1 or len(can_data) > 8:
951-
logger.warning(f"Uplink relay: invalid data in ref={ref}")
952-
return
876+
can_id = uplink_msg.get("canId")
877+
can_data = uplink_msg.get("data", [])
878+
ref = uplink_msg.get("ref", "unknown")
953879

954-
# Pack as uplink UDP packet: 0xCAFE + seq + count(1) + CAN message
955-
uplink_seq += 1
956-
data_bytes = bytes(can_data) + b'\x00' * (8 - len(can_data))
957-
can_msg = CANMessage(time.time(), can_id, data_bytes)
880+
if can_id is None or not isinstance(can_id, int) or can_id < 0:
881+
logger.warning(f"Uplink relay: invalid canId in ref={ref}")
882+
return
883+
if not isinstance(can_data, list) or len(can_data) < 1 or len(can_data) > 8:
884+
logger.warning(f"Uplink relay: invalid data in ref={ref}")
885+
return
958886

959-
payload = UPLINK_MAGIC
960-
payload += struct.pack("!QH", uplink_seq, 1)
961-
payload += can_msg.pack()
887+
# Pack as uplink UDP packet: 0xCAFE + seq + count(1) + CAN message
888+
uplink_seq += 1
889+
data_bytes = bytes(can_data) + b'\x00' * (8 - len(can_data))
890+
can_msg = CANMessage(time.time(), can_id, data_bytes)
962891

963-
try:
964-
uplink_sock.sendto(payload, (REMOTE_IP, UDP_PORT))
965-
logger.info(f"Uplink relayed to car: canId={can_id} ref={ref} seq={uplink_seq}")
966-
except (PermissionError, OSError) as e:
967-
logger.error(f"Uplink UDP send failed: {e}")
892+
payload = UPLINK_MAGIC
893+
payload += struct.pack("!QH", uplink_seq, 1)
894+
payload += can_msg.pack()
968895

969-
except Exception as e:
970-
logger.error(f"Uplink relay error: {e}")
896+
try:
897+
uplink_sock.sendto(payload, (REMOTE_IP, UDP_PORT))
898+
logger.info(f"Uplink relayed to car: canId={can_id} ref={ref} seq={uplink_seq}")
899+
except (PermissionError, OSError) as e:
900+
logger.error(f"Uplink UDP send failed: {e}")
971901

972-
await TelemetryNode._pump_pubsub_with_heartbeat(
973-
pubsub, r, _relay, log=logger,
974-
)
902+
except Exception as e:
903+
logger.error(f"Uplink relay error: {e}")
975904

976-
except Exception as e:
977-
logger.error(f"Uplink relay Redis error: {e}")
905+
# Reconnect loop: the pump returns whenever the pubsub connection
906+
# goes silent past HEARTBEAT_STALE_S, and we re-subscribe here.
907+
try:
908+
while True:
909+
r = None
910+
try:
911+
r = aioredis.from_url(REDIS_URL)
912+
pubsub = r.pubsub()
913+
await pubsub.subscribe(REDIS_UPLINK_CHANNEL, REDIS_HEARTBEAT_CHANNEL)
914+
logger.info(f"Subscribed to Redis channels: {REDIS_UPLINK_CHANNEL}, {REDIS_HEARTBEAT_CHANNEL}")
915+
await pump_pubsub_with_heartbeat(pubsub, _relay, log=logger)
916+
except asyncio.CancelledError:
917+
raise
918+
except Exception as e:
919+
logger.error(f"Uplink relay Redis error: {e}")
920+
await asyncio.sleep(1.0)
921+
finally:
922+
if r is not None:
923+
with contextlib.suppress(Exception):
924+
await r.aclose()
978925
finally:
979926
uplink_sock.close()
980927

@@ -1051,8 +998,8 @@ async def version_checker():
1051998
await asyncio.sleep(30.0)
1052999

10531000
# Base mode has a real Redis server; create an async client for the
1054-
# heartbeat writer. The writer writes telemetry:heartbeat every 1s so
1055-
# pubsub subscribers can detect a half-dead producer and reconnect.
1001+
# heartbeat writer. The writer publishes on the heartbeat channel every
1002+
# 1s so pubsub subscribers can detect a dead subscription and reconnect.
10561003
_async_redis = aioredis.from_url(REDIS_URL)
10571004
tasks = [udp_receiver(), missing_reporter(), stats_publisher(), raw_csv_logger(), car_time_injector(), version_checker(), utils.heartbeat_coro(self.telemetry_event), run_heartbeat_writer(_async_redis)]
10581005
if ENABLE_UPLINK:

‎universal-telemetry-software/src/heartbeat.py‎

Lines changed: 65 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,29 @@
11
"""
2-
Heartbeat writer — publishes a per-process liveness key to Redis every second
3-
so Redis pub/sub subscribers can detect a half-dead producer (TCP up, subscription
4-
state lost) and reconnect. See docs/superpowers/plans/2026-06-11-stack-resilience.md.
2+
Pubsub liveness probe — the producer publishes a heartbeat message on a Redis
3+
pubsub channel every second, and every subscriber also subscribes to that
4+
channel. Liveness is measured on the pubsub connection itself: if no message
5+
of any kind (heartbeat or data) arrives for HEARTBEAT_STALE_S, the connection
6+
is presumed half-dead and the subscriber tears down and re-subscribes.
7+
8+
An out-of-band check (e.g. GET on a heartbeat key) cannot detect this state:
9+
regular commands use a different pool connection that redis-py transparently
10+
reconnects, so the key would look fresh while the pubsub connection is dark.
11+
See docs/superpowers/plans/2026-06-11-stack-resilience.md.
512
"""
613
import asyncio
714
import json
815
import logging
916
import time
1017

11-
from .config import REDIS_HEARTBEAT_KEY
12-
from .redis_utils import get_async_client
18+
from .config import HEARTBEAT_STALE_S, REDIS_HEARTBEAT_CHANNEL
1319

1420
logger = logging.getLogger(__name__)
1521

1622
HEARTBEAT_INTERVAL_S = 1.0
1723

1824

1925
async def run_heartbeat_writer(redis_client=None) -> None:
20-
"""Write a heartbeat key to Redis every HEARTBEAT_INTERVAL_S.
26+
"""Publish a heartbeat message on REDIS_HEARTBEAT_CHANNEL every HEARTBEAT_INTERVAL_S.
2127
2228
Stops on cancellation. Logs and continues on any other exception so transient
2329
Redis blips don't take down the writer; if Redis is persistently down, the
@@ -30,10 +36,60 @@ async def run_heartbeat_writer(redis_client=None) -> None:
3036
while True:
3137
try:
3238
payload = json.dumps({"uptime_s": time.monotonic() - start_mono,
33-
"wall_ts": time.time()})
34-
await redis_client.set(REDIS_HEARTBEAT_KEY, payload, ex=30)
39+
"wall_ts": time.time()})
40+
await redis_client.publish(REDIS_HEARTBEAT_CHANNEL, payload)
3541
except asyncio.CancelledError:
3642
raise
3743
except Exception as e:
38-
logger.warning(f"Heartbeat write failed: {e}")
44+
logger.warning(f"Heartbeat publish failed: {e}")
3945
await asyncio.sleep(HEARTBEAT_INTERVAL_S)
46+
47+
48+
async def pump_pubsub_with_heartbeat(pubsub, on_message, *,
49+
stale_s: float = HEARTBEAT_STALE_S,
50+
should_stop=None,
51+
log=None):
52+
"""Drain a Redis pubsub, returning if the connection goes silent.
53+
54+
Replaces the naive `async for message in pubsub.listen():` pattern that
55+
silently goes dark when the TCP connection or subscription state is lost
56+
after a car power-cycle. The pubsub must be subscribed to
57+
REDIS_HEARTBEAT_CHANNEL in addition to its data channels; with the producer
58+
publishing every second, more than `stale_s` of silence means the
59+
subscription is dead, so the pump returns and the caller's outer loop
60+
re-subscribes. Heartbeat messages only refresh the liveness clock and are
61+
not forwarded to `on_message`.
62+
63+
`on_message` is an `async` callable invoked with the raw pubsub message
64+
dict for each non-heartbeat message. `should_stop` is an optional callable
65+
checked every iteration (pass `shutdown_event.is_set` for clean SIGTERM).
66+
Returns on staleness, pubsub errors, or should_stop; raises on cancellation.
67+
"""
68+
_log = log or logger
69+
last_msg_mono = time.monotonic() # armed at subscribe time
70+
while True:
71+
if should_stop is not None and should_stop():
72+
return
73+
try:
74+
msg = await pubsub.get_message(timeout=1.0, ignore_subscribe_messages=True)
75+
except asyncio.CancelledError:
76+
raise
77+
except Exception as e:
78+
_log.warning(f"pubsub.get_message error, reconnecting: {e}")
79+
await asyncio.sleep(0.5)
80+
return # let the outer while-True re-subscribe
81+
82+
if msg is not None:
83+
last_msg_mono = time.monotonic()
84+
channel = msg.get("channel")
85+
if isinstance(channel, bytes):
86+
channel = channel.decode("utf-8", errors="replace")
87+
if channel == REDIS_HEARTBEAT_CHANNEL:
88+
continue # liveness signal only — don't forward
89+
try:
90+
await on_message(msg)
91+
except Exception as e:
92+
_log.error(f"subscriber handler error: {e}")
93+
elif time.monotonic() - last_msg_mono > stale_s:
94+
_log.warning("heartbeat stale, forcing pubsub reconnect")
95+
return

‎universal-telemetry-software/src/websocket_bridge.py‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,11 @@
1515
REDIS_UPLINK_CHANNEL,
1616
REDIS_DIAG_CHANNEL,
1717
REDIS_WS_CLIENTS_KEY,
18+
REDIS_HEARTBEAT_CHANNEL,
1819
ENABLE_UPLINK,
1920
)
2021
from src import redis_utils, utils
21-
from src.data import TelemetryNode
22+
from src.heartbeat import pump_pubsub_with_heartbeat
2223

2324
logger = logging.getLogger("WebSocketBridge")
2425

@@ -298,13 +299,12 @@ async def redis_listener():
298299
try:
299300
r = redis.from_url(REDIS_URL, health_check_interval=30)
300301
pubsub = r.pubsub()
301-
await pubsub.subscribe(REDIS_CHANNEL, REDIS_STATS_CHANNEL, REDIS_DIAG_CHANNEL)
302-
logger.info(f"Subscribed to Redis channels: {REDIS_CHANNEL}, {REDIS_STATS_CHANNEL}, {REDIS_DIAG_CHANNEL}")
302+
await pubsub.subscribe(REDIS_CHANNEL, REDIS_STATS_CHANNEL, REDIS_DIAG_CHANNEL,
303+
REDIS_HEARTBEAT_CHANNEL)
304+
logger.info(f"Subscribed to Redis channels: {REDIS_CHANNEL}, {REDIS_STATS_CHANNEL}, {REDIS_DIAG_CHANNEL}, {REDIS_HEARTBEAT_CHANNEL}")
303305
delay = backoff_min # reset backoff once a subscribe succeeds
304306

305307
async def _handler(msg):
306-
if shutdown_event.is_set():
307-
return
308308
if msg['type'] != 'message':
309309
return
310310
data = redis_utils.decode_message(msg['data'])
@@ -317,7 +317,8 @@ async def _handler(msg):
317317
return_exceptions=True
318318
)
319319

320-
await TelemetryNode._pump_pubsub_with_heartbeat(pubsub, r, _handler, log=logger)
320+
await pump_pubsub_with_heartbeat(pubsub, _handler,
321+
should_stop=shutdown_event.is_set, log=logger)
321322
except asyncio.CancelledError:
322323
raise
323324
except Exception as e:

0 commit comments

Comments
 (0)