Navidrome: self-heal the connection instead of latching disconnected (jimmydotcom)

A transient ping failure (network blip, Navidrome busy mid-scan) makes
_setup_client null out the configured creds, and _connection_attempted then
latches the client "disconnected" — so is_connected() returned False forever until
the user hit the manual Test button to re-read config. That's the reported
"disconnects every 5-10 min, reconnects instantly on Test."

Fix: ensure_connection no longer latches on a failed attempt — once a short
throttle (_RECONNECT_THROTTLE_S = 20s) elapses it re-attempts, and is_connected()
triggers that retry whenever it's currently disconnected. So a blip recovers on its
own within the next status check, no manual reconnect. The throttle prevents ping
storms when Navidrome is genuinely down.

Tests: transient failure self-heals after the throttle (and doesn't re-ping within
it); a connected client never re-pings; first connect attempts once. 115 navidrome/
media-server tests green.
This commit is contained in:
BoulderBadgeDad 2026-06-11 21:34:48 -07:00
parent d9cda0c31c
commit 4dd09ff48a
2 changed files with 81 additions and 7 deletions

View file

@ -1,6 +1,7 @@
import requests
import hashlib
import secrets
import time
from typing import List, Optional, Dict, Any
from datetime import datetime
from urllib.parse import urlencode
@ -161,6 +162,7 @@ class NavidromeClient(MediaServerClient):
self.music_folder_id: Optional[str] = None
self._connection_attempted = False
self._is_connecting = False
self._last_connect_attempt = 0.0 # monotonic time of the last connect try
# Cache for performance
self._artist_cache = {}
@ -245,15 +247,32 @@ class NavidromeClient(MediaServerClient):
logger.error(f"Error setting music folder: {e}")
return False
# A failed connect used to latch the client "disconnected" until the user hit
# the manual Test button (a transient ping failure nukes the creds in
# _setup_client). Re-attempt at most this often so it self-heals on its own.
_RECONNECT_THROTTLE_S = 20.0
def ensure_connection(self) -> bool:
"""Ensure connection to Navidrome server with lazy initialization."""
if self._connection_attempted:
return self.base_url is not None and self.username is not None
"""Ensure connection to Navidrome with lazy init + self-healing retry.
Already connected return True. A prior FAILED attempt no longer latches
forever: once _RECONNECT_THROTTLE_S has elapsed it re-attempts, so a
transient ping failure (network blip, Navidrome busy mid-scan) recovers by
itself instead of needing the manual "Test" reconnect."""
if self.base_url is not None and self.username is not None:
return True
if self._is_connecting:
return False
# Disconnected but attempted recently → don't hammer; let it heal on the
# next check past the throttle window.
if self._connection_attempted and \
(time.monotonic() - self._last_connect_attempt) < self._RECONNECT_THROTTLE_S:
return False
self._is_connecting = True
self._last_connect_attempt = time.monotonic()
try:
self._setup_client()
return self.base_url is not None and self.username is not None
@ -462,10 +481,11 @@ class NavidromeClient(MediaServerClient):
return None
def is_connected(self) -> bool:
"""Check if connected to Navidrome server"""
if not self._connection_attempted:
if not self._is_connecting:
self.ensure_connection()
"""Connected = configured + last connect OK. When NOT connected, trigger a
(throttled) reconnect attempt so the status self-heals instead of staying
latched disconnected until a manual Test."""
if not (self.base_url and self.username and self.password) and not self._is_connecting:
self.ensure_connection()
return (self.base_url is not None and
self.username is not None and
self.password is not None)

View file

@ -0,0 +1,54 @@
"""Navidrome connection self-heals after a transient ping failure (jimmydotcom).
A failed ping nukes the creds in _setup_client; previously _connection_attempted
latched the client 'disconnected' until a manual Test. Now is_connected re-attempts
(throttled) so a blip recovers on its own."""
from __future__ import annotations
from core.navidrome_client import NavidromeClient
def test_self_heals_after_transient_failure(monkeypatch):
c = NavidromeClient()
calls = {'n': 0}
def fake_setup():
calls['n'] += 1
if calls['n'] == 1: # transient failure nukes creds
c.base_url = c.username = c.password = None
else: # blip passed → connects
c.base_url, c.username, c.password = 'http://nd:4533', 'u', 'p'
monkeypatch.setattr(c, '_setup_client', fake_setup)
assert c.is_connected() is False # first attempt fails
assert calls['n'] == 1
assert c.is_connected() is False # immediate recheck: throttled
assert calls['n'] == 1 # did NOT re-ping (no storm)
c._last_connect_attempt -= (c._RECONNECT_THROTTLE_S + 1) # throttle window elapses
assert c.is_connected() is True # re-attempts → recovers itself
assert calls['n'] == 2 # no manual reconnect was needed
def test_connected_client_does_not_reattempt(monkeypatch):
c = NavidromeClient()
c.base_url, c.username, c.password = 'http://nd', 'u', 'p'
c._connection_attempted = True
calls = {'n': 0}
monkeypatch.setattr(c, '_setup_client', lambda: calls.__setitem__('n', calls['n'] + 1))
assert c.is_connected() is True
assert calls['n'] == 0 # already connected → never re-pings
def test_first_connect_attempts_once(monkeypatch):
c = NavidromeClient()
calls = {'n': 0}
def fake_setup():
calls['n'] += 1
c.base_url, c.username, c.password = 'http://nd', 'u', 'p'
monkeypatch.setattr(c, '_setup_client', fake_setup)
assert c.is_connected() is True
assert calls['n'] == 1