Skip to content

Commit de57df2

Browse files
mdrxymdrxy
committed
feat: Enhance override management and notification handling in failsafe.py
1 parent 0176e4a commit de57df2

1 file changed

Lines changed: 181 additions & 73 deletions

File tree

‎failsafe.py‎

Lines changed: 181 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,11 @@ def __init__(self) -> None:
146146
self.last_healthcheck_time: datetime | None = None
147147
self.healthcheck_failures = 0
148148
self.max_healthcheck_failures = 5
149+
self.state_changed_during_override = False
150+
self.pending_source_change: tuple[str, str] | None = None
151+
self.override_logged = False
152+
self.original_source_before_override: str | None = None
153+
self.last_healthcheck_retry_time: datetime | None = None
149154

150155

151156
# Global state manager for temporary override functionality
@@ -245,6 +250,10 @@ def handle_override_message(message: dict[str, Any]) -> None:
245250
override_manager.end_time = datetime.now(timezone.utc) + timedelta(
246251
minutes=duration_minutes
247252
)
253+
override_manager.state_changed_during_override = False
254+
override_manager.pending_source_change = None
255+
override_manager.override_logged = False
256+
override_manager.original_source_before_override = None
248257

249258
logger.info(
250259
"Temporary override activated for %d minutes (until %s UTC)",
@@ -255,6 +264,10 @@ def handle_override_message(message: dict[str, Any]) -> None:
255264
elif action == "disable_override":
256265
override_manager.active = False
257266
override_manager.end_time = None
267+
override_manager.state_changed_during_override = False
268+
override_manager.pending_source_change = None
269+
override_manager.override_logged = False
270+
override_manager.original_source_before_override = None
258271
logger.info("Temporary override manually disabled")
259272

260273
else:
@@ -264,42 +277,62 @@ def handle_override_message(message: dict[str, Any]) -> None:
264277
logger.exception("Error processing override message")
265278

266279

267-
def check_override_expiry() -> None:
280+
def check_override_expiry() -> bool:
268281
"""Check if the temporary override has expired and disable it if so.
269282
270283
Compares current time with override end time and resets override state if the period
271284
has elapsed.
285+
286+
Returns:
287+
True if override just expired, False otherwise
272288
"""
273289
if override_manager.active and override_manager.end_time:
274290
current_time = datetime.now(timezone.utc)
275291
if current_time >= override_manager.end_time:
276292
override_manager.active = False
277293
override_manager.end_time = None
294+
override_manager.override_logged = False
278295
logger.info(
279296
"Temporary override expired/disabled: returning to normal operation"
280297
)
298+
return True
299+
return False
281300

282301

283302
def send_health_check(publishers: dict[str, RabbitMQPublisher | None]) -> None:
284303
"""Send a health check message to RabbitMQ indicating the system is alive.
285304
286305
Publishes a heartbeat message with current system status including pin state, active
287-
source, and override status. Stops attempting health checks after max failures.
306+
source, and override status. After max failures, retries every hour to allow
307+
recovery if RabbitMQ comes back online.
288308
"""
289309
healthcheck_publisher = publishers.get("healthcheck")
290310
if not healthcheck_publisher:
291311
return
292312

293-
# Stop attempting health checks if we've exceeded max failures
313+
current_time = datetime.now(timezone.utc)
314+
315+
# If we've exceeded max failures, only retry every hour
294316
if (
295317
override_manager.healthcheck_failures
296318
>= override_manager.max_healthcheck_failures
297319
):
298-
return
299-
300-
current_time = datetime.now(timezone.utc)
320+
# Check if it's time for an hourly retry
321+
if (
322+
override_manager.last_healthcheck_retry_time is None
323+
or current_time - override_manager.last_healthcheck_retry_time
324+
>= timedelta(hours=1)
325+
):
326+
logger.info(
327+
"Attempting hourly health check retry after %d failures",
328+
override_manager.healthcheck_failures,
329+
)
330+
override_manager.last_healthcheck_retry_time = current_time
331+
# Continue to attempt sending below
332+
else:
333+
return # Skip if not time for retry yet
301334

302-
# Send health check every hour
335+
# Send health check every hour (or during retry attempts)
303336
if (
304337
override_manager.last_healthcheck_time is None
305338
or current_time - override_manager.last_healthcheck_time >= timedelta(hours=1)
@@ -322,23 +355,42 @@ def send_health_check(publishers: dict[str, RabbitMQPublisher | None]) -> None:
322355
app_config.healthcheck_exchange.routing_keys["health_ping"], health_payload
323356
):
324357
override_manager.last_healthcheck_time = current_time
358+
359+
# If this was a successful retry after failures, log recovery
360+
if (
361+
override_manager.healthcheck_failures
362+
>= override_manager.max_healthcheck_failures
363+
):
364+
logger.info(
365+
"Health check publishing recovered after %d failures. "
366+
"RabbitMQ connection restored.",
367+
override_manager.healthcheck_failures,
368+
)
369+
325370
override_manager.healthcheck_failures = 0 # Reset on success
326371
logger.info("Health check message sent successfully")
327372
else:
328-
override_manager.healthcheck_failures += 1
373+
# Only increment failures if we haven't reached max yet
374+
# (to avoid inflating the counter during hourly retries)
375+
if (
376+
override_manager.healthcheck_failures
377+
< override_manager.max_healthcheck_failures
378+
):
379+
override_manager.healthcheck_failures += 1
380+
329381
logger.error(
330382
"Failed to send health check message (attempt %d/%d)",
331383
override_manager.healthcheck_failures,
332384
override_manager.max_healthcheck_failures,
333385
)
386+
334387
if (
335388
override_manager.healthcheck_failures
336389
>= override_manager.max_healthcheck_failures
337390
):
338391
logger.warning(
339-
"Maximum health check failures reached (%d). Disabling health "
340-
"check messages to prevent infinite retries. Check RabbitMQ "
341-
"configuration.",
392+
"Maximum health check failures reached (%d). Will retry every hour "
393+
"until RabbitMQ connection is restored.",
342394
override_manager.max_healthcheck_failures,
343395
)
344396

@@ -898,6 +950,60 @@ def send_groupme_notification(
898950
)
899951

900952

953+
def send_all_source_change_notifications(
954+
current_source: str,
955+
prev_source: str,
956+
*,
957+
current_pin_state: bool,
958+
local_rabbitmq_publishers: dict[str, RabbitMQPublisher | None],
959+
) -> None:
960+
"""Send all configured notifications for a source change."""
961+
playlist_data: dict[str, Any] | None = None
962+
persona_data: dict[str, Any] | None = None
963+
964+
# Only fetch Spinitron if configured
965+
if app_config.spinitron_api_base_url:
966+
playlist_data = get_current_playlist()
967+
if playlist_data:
968+
persona_data = resolve_and_notify_dj(playlist_data, current_source)
969+
970+
# Send Discord Source Change Notification
971+
if app_config.discord_webhook_url:
972+
send_discord_source_change(current_source, playlist_data, persona_data)
973+
974+
# Send GroupMe Management Notification
975+
if app_config.groupme_bot_id_mgmt:
976+
send_groupme_notification(
977+
current_source,
978+
app_config.groupme_bot_id_mgmt,
979+
is_public_dj_alert=False,
980+
)
981+
982+
# Send RabbitMQ notification
983+
notifications_publisher = local_rabbitmq_publishers.get("notifications")
984+
if notifications_publisher:
985+
rabbitmq_payload = {
986+
"source_application": "wbor-failsafe-notifier",
987+
"event_type": "source_change",
988+
"timestamp_utc": datetime.now(timezone.utc).isoformat(),
989+
"pin_name": PIN_NAME,
990+
"current_pin_state": current_pin_state,
991+
"active_source": current_source,
992+
"previous_active_source": prev_source,
993+
"details": {
994+
"playlist": playlist_data if playlist_data else {},
995+
"persona": persona_data if persona_data else {},
996+
},
997+
}
998+
routing_key = app_config.notifications_exchange.routing_keys["source_change"]
999+
logger.info(
1000+
"Publishing source change event to RabbitMQ with routing key `%s`.",
1001+
routing_key,
1002+
)
1003+
if not notifications_publisher.publish(routing_key, rabbitmq_payload):
1004+
logger.error("Failed to publish source change event to RabbitMQ.")
1005+
1006+
9011007
def main_loop(
9021008
local_rabbitmq_publishers: dict[str, RabbitMQPublisher | None],
9031009
local_rabbitmq_consumer: RabbitMQConsumer | None,
@@ -922,32 +1028,25 @@ def main_loop(
9221028
)
9231029

9241030
# Start the RabbitMQ consumer if available
1031+
# (Used for override commands)
9251032
if local_rabbitmq_consumer:
9261033
if local_rabbitmq_consumer.start_consuming():
9271034
logger.info("RabbitMQ consumer started successfully")
9281035
else:
9291036
logger.error("Failed to start RabbitMQ consumer")
9301037

931-
# Wait for the pin to change state
1038+
# The core of the app: wait for the pin to change state
9321039
while True:
9331040
try:
934-
# Check for override expiry
935-
check_override_expiry()
1041+
# Check for override expiry and handle pending notifications
1042+
override_just_expired = check_override_expiry()
9361043

937-
# Send health check if needed
9381044
send_health_check(local_rabbitmq_publishers)
9391045

9401046
current_pin_state = DIGITAL_PIN.value
9411047
current_source = PRIMARY_SOURCE if current_pin_state else BACKUP_SOURCE
9421048

943-
# Check if we should skip processing due to active override
944-
if override_manager.active:
945-
logger.debug("Override is active, skipping state change processing")
946-
prev_pin_state = current_pin_state
947-
prev_source = current_source
948-
time.sleep(0.5)
949-
continue
950-
1049+
# Always detect and log state changes
9511050
if current_pin_state != prev_pin_state:
9521051
logger.info(
9531052
"Source changed from `%s` (pin: %s) to `%s` (pin: %s)",
@@ -957,63 +1056,68 @@ def main_loop(
9571056
current_pin_state,
9581057
)
9591058

960-
playlist_data: dict[str, Any] | None = None
961-
persona_data: dict[str, Any] | None = None
1059+
if override_manager.active:
1060+
# Capture original source when override first becomes active
1061+
if override_manager.original_source_before_override is None:
1062+
override_manager.original_source_before_override = prev_source
9621063

963-
# Only fetch Spinitron if configured
964-
if app_config.spinitron_api_base_url:
965-
playlist_data = get_current_playlist()
966-
if playlist_data:
967-
persona_data = resolve_and_notify_dj(
968-
playlist_data, current_source
1064+
# Log once that override is active during state change
1065+
if not override_manager.override_logged:
1066+
logger.info(
1067+
"Override is active, notifications suppressed for this "
1068+
"state change"
9691069
)
970-
971-
# Send Discord Source Change Notification
972-
if app_config.discord_webhook_url:
973-
send_discord_source_change(
974-
current_source, playlist_data, persona_data
1070+
override_manager.override_logged = True
1071+
# Track that state changed during override
1072+
override_manager.state_changed_during_override = True
1073+
override_manager.pending_source_change = (
1074+
prev_source,
1075+
current_source,
9751076
)
976-
977-
# Send GroupMe Management Notification
978-
if app_config.groupme_bot_id_mgmt:
979-
send_groupme_notification(
1077+
else:
1078+
# Send notifications normally when override is not active
1079+
send_all_source_change_notifications(
9801080
current_source,
981-
app_config.groupme_bot_id_mgmt,
982-
is_public_dj_alert=False,
1081+
prev_source,
1082+
current_pin_state=current_pin_state,
1083+
local_rabbitmq_publishers=local_rabbitmq_publishers,
9831084
)
9841085

985-
notifications_publisher = local_rabbitmq_publishers.get("notifications")
986-
if notifications_publisher:
987-
rabbitmq_payload = {
988-
"source_application": "wbor-failsafe-notifier",
989-
"event_type": "source_change",
990-
"timestamp_utc": datetime.now(timezone.utc).isoformat(),
991-
"pin_name": PIN_NAME,
992-
"current_pin_state": current_pin_state,
993-
"active_source": current_source,
994-
"previous_active_source": prev_source,
995-
"details": {
996-
"playlist": playlist_data if playlist_data else {},
997-
"persona": persona_data if persona_data else {},
998-
},
999-
}
1000-
routing_key = app_config.notifications_exchange.routing_keys[
1001-
"source_change"
1002-
]
1086+
# Handle notifications after override expires
1087+
elif (
1088+
override_just_expired and override_manager.state_changed_during_override
1089+
):
1090+
# Only send notification if current source differs from original source
1091+
if (
1092+
override_manager.original_source_before_override is not None
1093+
and current_source
1094+
!= override_manager.original_source_before_override
1095+
):
10031096
logger.info(
1004-
"Publishing source change event to RabbitMQ with routing key "
1005-
"`%s`.",
1006-
routing_key,
1097+
"Override expired, sending delayed notification for source "
1098+
"change from `%s` to `%s`",
1099+
override_manager.original_source_before_override,
1100+
current_source,
10071101
)
1008-
if not notifications_publisher.publish(
1009-
routing_key, rabbitmq_payload
1010-
):
1011-
logger.error(
1012-
"Failed to publish source change event to RabbitMQ."
1013-
)
1102+
send_all_source_change_notifications(
1103+
current_source,
1104+
override_manager.original_source_before_override,
1105+
current_pin_state=current_pin_state,
1106+
local_rabbitmq_publishers=local_rabbitmq_publishers,
1107+
)
1108+
else:
1109+
logger.info(
1110+
"Override expired, but source returned to original state (%s), "
1111+
"no notification needed",
1112+
override_manager.original_source_before_override or "unknown",
1113+
)
1114+
override_manager.state_changed_during_override = False
1115+
override_manager.pending_source_change = None
1116+
override_manager.original_source_before_override = None
10141117

1015-
prev_pin_state = current_pin_state
1016-
prev_source = current_source
1118+
# Update state tracking
1119+
prev_pin_state = current_pin_state
1120+
prev_source = current_source
10171121

10181122
time.sleep(0.5) # Check interval for pin state
10191123

@@ -1032,14 +1136,18 @@ def main() -> None:
10321136
local_rabbitmq_consumer: RabbitMQConsumer | None = None
10331137
try:
10341138
if DRY_RUN:
1035-
logger.info("Starting WBOR Failsafe Notifier in DRY_RUN mode...")
1036-
logger.info("Configuration validation successful")
1139+
logger.info("Starting WBOR Failsafe Notifier in DRY_RUN mode (CI)...")
10371140
logger.info(
10381141
"Primary Source: `%s`, Backup Source: `%s`, Monitored Pin: `%s`",
10391142
PRIMARY_SOURCE,
10401143
BACKUP_SOURCE,
10411144
PIN_NAME,
10421145
)
1146+
1147+
# Don't initialize RabbitMQ or GPIO in DRY_RUN mode
1148+
logger.info("Skipping RabbitMQ and GPIO initialization in DRY_RUN mode.")
1149+
logger.info("Skipping main loop in DRY_RUN mode.")
1150+
logger.info("WBOR Failsafe Notifier DRY_RUN mode complete.")
10431151
return
10441152

10451153
logger.info("Starting WBOR Failsafe Notifier...")

0 commit comments

Comments
 (0)