Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/architecture/network.md
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ Interface SRT / RTSP / RTP reach the camera through, **by name**; empty follows

Resolving a name the operator configured as a destination is a query to the show LAN's own resolver, never an outbound connection. The Interface Assignment panel and the bundle's A5 / A6 ask the same shared bounded resolvers the outputs use (`net_utils.HOST_RESOLVER` / `IPV4_RESOLVER`), only for names configured as an output's destination, waiting at most 0.2 s per render. **A caller waits its own budget counted from when the lookup started**, so a short wait (the panel's first, zero-wait pass) never shortens another caller's.

An OSC destination whose client cannot be created (a name that does not resolve, a refused interface pin, a malformed host) waits `_CREATE_RETRY_S` before `OscService` tries it again, pinned or not, so a row streaming at its frame rate never looks it up or logs per send. Its failures log one WARNING when they start, one reminder at most every `_CREATE_FAILURE_REMINDER_S` with the attempt count, and one INFO when its client is created. An interface eviction (`evict_egress` / `retain_egress`) clears a pinned target's retry delay but not its streak, so a target that keeps failing across an address change is reminded, not announced again.

## Discovery (`web/discovery.py`)
- UDP multicast `239.255.50.50:50505` (`BEACON_MCAST_GROUP` / `BEACON_PORT`), JSON beacon every `BEACON_INTERVAL` = 2s
- `iface_ip` used for both `IP_MULTICAST_IF` (send) and `IP_ADD_MEMBERSHIP` (receive)
Expand Down
74 changes: 58 additions & 16 deletions openfollow/osc/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,10 +49,13 @@
# uniform 4-tuple across transports (framing is a TCP-only concern).
_UDP_FRAMING_PIN = "length_prefix"

# How long a pinned target whose client could not be created waits before the
# next attempt. The address a pin needs can lag the interface by an observer
# debounce, so the failure is expected to clear on its own.
_PINNED_CREATE_RETRY_S = 1.0
# How long a target whose client could not be created waits before the next
# attempt, so a name that does not resolve, or a refused pin, is not retried per
# send. The address a pin needs can lag the interface by an observer debounce,
# so that failure is expected to clear on its own.
_CREATE_RETRY_S = 1.0
# How often a target whose client still cannot be created is logged again.
_CREATE_FAILURE_REMINDER_S = 60.0

# One cached client per target and the interface it is pinned to.
_CacheKey = tuple[str, int, str, str, Egress | None]
Expand Down Expand Up @@ -194,6 +197,14 @@ class ClientStats:
last_error: str = ""


@dataclass
class _CreateFailure:
"""A target's run of failed client creations, for its throttled log lines."""

logged_at: float
attempts: int = 1


@dataclass
class _ClientEntry:
"""Cached OSC client for a single ``(host, port, protocol, framing)`` target.
Expand Down Expand Up @@ -328,9 +339,10 @@ class OscService:
def __init__(self, *, close_async: Callable[[list[Any]], None] = _close_clients_in_background) -> None:
self._cache: dict[_CacheKey, _ClientEntry] = {}
self._cache_lock = threading.Lock()
# Pinned targets whose client could not be created, and when to try
# again: a refused pin would otherwise be retried and logged per send.
# Targets whose client could not be created, and when to try again:
# each would otherwise be retried and logged per send.
self._create_retry_at: dict[_CacheKey, float] = {}
self._create_failures: dict[_CacheKey, _CreateFailure] = {}
self._close_async = close_async

self._dispatcher: Any = _GuardedDispatcher() if _PYTHONOSC_AVAILABLE else None
Expand Down Expand Up @@ -515,18 +527,15 @@ def _get_or_create_client(
except (OSError, ValueError) as exc:
# OSError: socket/DNS failure or a refused pin. ValueError:
# malformed host string.
logger.error(
"Failed to create OSC client for %s://%s:%d – %s",
protocol,
host,
port,
exc,
)
if egress is not None:
with self._cache_lock:
self._create_retry_at[key] = time.monotonic() + _PINNED_CREATE_RETRY_S
self._record_create_failure(key, exc)
return None
new_entry = _ClientEntry(client=client)
with self._cache_lock:
failure = self._create_failures.pop(key, None)
if failure is not None:
logger.info(
"Created OSC client for %s://%s:%d after %d failed attempts", protocol, host, port, failure.attempts
)
with self._cache_lock:
# Re-check under lock to resolve a duplicate-creation race.
existing = self._cache.get(key)
Expand All @@ -537,6 +546,39 @@ def _get_or_create_client(
self._cache[key] = new_entry
return new_entry

def _record_create_failure(self, key: _CacheKey, exc: Exception) -> None:
"""Hold *key* back for the retry delay; log a streak's start, then once a minute."""
host, port, protocol = key[0], key[1], key[2]
now = time.monotonic()
with self._cache_lock:
self._create_retry_at[key] = now + _CREATE_RETRY_S
failure = self._create_failures.get(key)
if failure is None:
self._create_failures[key] = _CreateFailure(logged_at=now)
else:
failure.attempts += 1
if now - failure.logged_at < _CREATE_FAILURE_REMINDER_S:
return
failure.logged_at = now
if failure is None:
logger.warning(
"Cannot create OSC client for %s://%s:%d, retrying every %gs: %s",
protocol,
host,
port,
_CREATE_RETRY_S,
exc,
)
else:
logger.warning(
"Still cannot create OSC client for %s://%s:%d after %d attempts: %s",
protocol,
host,
port,
failure.attempts,
exc,
)

# ------------------------------------------------------------------
# Inbound
# ------------------------------------------------------------------
Expand Down
155 changes: 153 additions & 2 deletions tests/test_osc_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,13 +216,164 @@ def test_send_with_client_construction_error_returns_silently(
monkeypatch.setattr(service_module, "SimpleUDPClient", _ExplodingClient)
monkeypatch.setattr(service_module, "_resolve_host", lambda host: host)
svc = OscService()
with caplog.at_level(logging.ERROR):
with caplog.at_level(logging.WARNING):
svc.send("/a", host="bogus", port=1)
assert any("Failed to create OSC client" in r.message for r in caplog.records)
assert [r.getMessage() for r in caplog.records] == [
"Cannot create OSC client for udp://bogus:1, retrying every 1s: DNS failure for bogus:1"
]
# No entry cached because client construction blew up.
assert svc.stats_for("bogus", 1) == ClientStats()


class TestClientCreationFailure:
"""A target whose client cannot be created waits before retrying and logs once per failure streak."""

_DNS_FAILURE = "DNS lookup for 'console.invalid' failed: Name or service not known"

@pytest.fixture
def unresolvable(self, monkeypatch: pytest.MonkeyPatch) -> tuple[OscService, list[float], list[str], list[bool]]:
"""A service whose clock stands still unless a test moves it, and whose lookups fail until told otherwise."""
from openfollow.net_utils import HostLookup

now = [100.0]
looked_up: list[str] = []
resolves = [False]

def _lookup(host: str, _wait_s: float) -> HostLookup:
looked_up.append(host)
if resolves[0]:
return HostLookup("resolved", ("192.0.2.40",))
return HostLookup("failed", error="Name or service not known")

monkeypatch.setattr(service_module, "time", SimpleNamespace(monotonic=lambda: now[0]))
monkeypatch.setattr(service_module, "_resolver", SimpleNamespace(lookup=_lookup))
monkeypatch.setattr(service_module, "SimpleUDPClient", _FakeClient)
return OscService(), now, looked_up, resolves

@staticmethod
def _send_for(svc: OscService, now: list[float], seconds: float, *, step: float = 0.25) -> list[bool]:
"""Send to console.invalid every *step* seconds for *seconds*."""
sent = []
for _ in range(round(seconds / step)):
sent.append(svc.send("/x", host="console.invalid", port=8000))
now[0] += step
return sent

def test_a_name_that_does_not_resolve_is_tried_once_per_retry_and_logged_once(
self,
unresolvable: tuple[OscService, list[float], list[str], list[bool]],
caplog: pytest.LogCaptureFixture,
) -> None:
"""A row streaming at its frame rate would otherwise look it up and log per send."""
svc, now, looked_up, _resolves = unresolvable
with caplog.at_level(logging.INFO, logger="openfollow"):
sent = [svc.send("/x", host="console.invalid", port=8000) for _ in range(600)]
now[0] += 1.0
sent += [svc.send("/x", host="console.invalid", port=8000) for _ in range(600)]
assert not any(sent)
assert looked_up == ["console.invalid"] * 2
assert [(r.levelno, r.getMessage()) for r in caplog.records] == [
(
logging.WARNING,
f"Cannot create OSC client for udp://console.invalid:8000, retrying every 1s: {self._DNS_FAILURE}",
),
]

def test_a_streak_is_reminded_once_a_minute_and_its_recovery_logged(
self,
unresolvable: tuple[OscService, list[float], list[str], list[bool]],
caplog: pytest.LogCaptureFixture,
) -> None:
svc, now, _looked_up, resolves = unresolvable
with caplog.at_level(logging.INFO, logger="openfollow"):
self._send_for(svc, now, 150.0)
resolves[0] = True
assert self._send_for(svc, now, 2.0)[-1] is True
still = "Still cannot create OSC client for udp://console.invalid:8000 after"
assert [(r.levelno, r.getMessage()) for r in caplog.records] == [
(
logging.WARNING,
f"Cannot create OSC client for udp://console.invalid:8000, retrying every 1s: {self._DNS_FAILURE}",
),
(logging.WARNING, f"{still} 61 attempts: {self._DNS_FAILURE}"),
(logging.WARNING, f"{still} 121 attempts: {self._DNS_FAILURE}"),
(logging.INFO, "Created OSC client for udp://console.invalid:8000 after 150 failed attempts"),
]
assert _FakeClient.instances[0].host == "192.0.2.40"

def test_a_streak_after_a_recovery_starts_afresh(
self,
unresolvable: tuple[OscService, list[float], list[str], list[bool]],
caplog: pytest.LogCaptureFixture,
) -> None:
svc, now, _looked_up, resolves = unresolvable
with caplog.at_level(logging.INFO, logger="openfollow"):
self._send_for(svc, now, 2.0)
resolves[0] = True
self._send_for(svc, now, 2.0)
svc.evict("console.invalid", 8000)
resolves[0] = False
self._send_for(svc, now, 2.0)
started = f"Cannot create OSC client for udp://console.invalid:8000, retrying every 1s: {self._DNS_FAILURE}"
assert [r.getMessage() for r in caplog.records] == [
started,
"Created OSC client for udp://console.invalid:8000 after 2 failed attempts",
started,
]

def test_an_eviction_retries_at_once_but_keeps_the_streak(
self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""An address change is no new failure, so it is not announced again."""
attempts: list[str] = []

def _fail(host: str, *_a: Any, **_k: Any) -> Any:
attempts.append(host)
raise InterfaceUnavailable("cannot send via eth1")

monkeypatch.setattr(service_module, "time", SimpleNamespace(monotonic=lambda: 100.0))
monkeypatch.setattr(service_module, "_make_client", _fail)
egress = Egress("eth1", "198.51.100.10")
svc = OscService(close_async=lambda _clients: None)
with caplog.at_level(logging.INFO, logger="openfollow"):
svc.send("/x", host="198.51.100.20", port=9000, egress=egress)
svc.evict_egress("eth1")
svc.send("/x", host="198.51.100.20", port=9000, egress=egress)
assert len(attempts) == 2
assert [r.getMessage() for r in caplog.records] == [
"Cannot create OSC client for udp://198.51.100.20:9000, retrying every 1s: cannot send via eth1"
]

@pytest.mark.parametrize(
("failure", "egress"),
[
(OSError("DNS lookup for '198.51.100.20' failed"), None),
(InterfaceUnavailable("cannot send via eth1"), Egress("eth1", "198.51.100.10")),
(ValueError("bad host string"), None),
],
ids=["unresolvable", "refused-pin", "malformed-host"],
)
def test_every_failed_creation_waits_before_retrying(
self, monkeypatch: pytest.MonkeyPatch, failure: Exception, egress: Egress | None
) -> None:
now = [100.0]
attempts: list[str] = []

def _fail(host: str, *_a: Any, **_k: Any) -> Any:
attempts.append(host)
raise failure

monkeypatch.setattr(service_module, "time", SimpleNamespace(monotonic=lambda: now[0]))
monkeypatch.setattr(service_module, "_make_client", _fail)
svc = OscService()
for _ in range(5):
assert svc.send("/x", host="198.51.100.20", port=9000, egress=egress) is False
assert len(attempts) == 1
now[0] += 1.0
svc.send("/x", host="198.51.100.20", port=9000, egress=egress)
assert len(attempts) == 2


def test_send_message_failure_increments_error_counter(
patched_service: OscService,
caplog: pytest.LogCaptureFixture,
Expand Down
Loading