feat: reuse owned HTTPS sessions in ordinary shadow learning

Author: Maksym Shtembuliak

CHANGELOG.md

@@ -9,6 +9,12 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Added
 
+- SmartESS and DESSMonitor active Shadow Learning reuse one verified HTTPS
+  connection per control-discovery run. Unknown-delivery failures stop the run
+  without retrying commands; cancellation drains the outstanding request before
+  cleanup. Safe phase diagnostics omit signed URLs and credentials. Read-only
+  preparation, ValueCloud and ordinary polling are unchanged.
+
 - Active control learning shares numeric command evidence across SmartESS,
   DESSMonitor and generic ValueCloud controls. Missing cloud ranges no longer
   exclude numeric discovery; two intercepted values must agree on the supported

custom_components/eybond_local/cloud_http_session.py

@@ -0,0 +1,131 @@
+"""Operation-owned HTTPS reuse; never retry a semantic cloud write.
+
+No credentials, redirects, global pool or provider fallback. The owner drains
+its blocking worker before closing this resource. Construction performs no IO.
+"""
+from __future__ import annotations
+
+from http.client import HTTPSConnection, IncompleteRead
+from io import BytesIO
+import logging
+from math import isfinite
+from socket import gaierror
+from ssl import SSLError
+from threading import Lock
+from time import monotonic
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlsplit
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _failure_reason(exc: BaseException) -> str:
+    """Closed diagnostics only: never stringify provider/socket exceptions."""
+    for kind, reason in ((TimeoutError, "timeout"), (gaierror, "dns"),
+                         (SSLError, "tls"), (HTTPError, "http_status"),
+                         (IncompleteRead, "incomplete_response"), (OSError, "network"),
+                         (ValueError, "invalid_response")):
+        if isinstance(exc, kind):
+            return reason
+    return "unexpected"
+
+
+class CloudHttpSession:
+    """A single origin/path, serialized requests, terminal failure/close."""
+
+    def __init__(self, base_url: str):
+        parsed = self._parse(base_url)
+        if parsed.query:
+            raise ValueError("cloud_http_origin_invalid")
+        self._origin = (parsed.hostname, parsed.port or 443)
+        self._path = parsed.path or "/"
+        self._connection = None
+        self._closed = False
+        self._lock = Lock()
+
+    @staticmethod
+    def _parse(url):
+        if type(url) is not str or any(ord(c) <= 32 or ord(c) >= 127 for c in url):
+            raise ValueError("cloud_http_url_invalid")
+        parsed = urlsplit(url)
+        if (parsed.scheme != "https" or not parsed.hostname or parsed.username is not None
+                or parsed.password is not None or parsed.fragment):
+            raise ValueError("cloud_http_origin_invalid")
+        return parsed
+
+    def get(self, url: str, *, headers: dict[str, str], timeout: float,
+            max_bytes: int = 2 * 1024 * 1024) -> bytes:
+        if not self._lock.acquire(blocking=False):
+            raise RuntimeError("cloud_http_busy")
+        phase, reused, started = "validation", False, monotonic()
+        try:
+            if self._closed:
+                raise ValueError("cloud_http_closed")
+            parsed = self._parse(url)
+            if ((parsed.hostname, parsed.port or 443) != self._origin
+                    or (parsed.path or "/") != self._path):
+                raise ValueError("cloud_http_origin_mismatch")
+            if (type(timeout) not in (int, float) or not isfinite(timeout) or timeout <= 0
+                    or type(max_bytes) is not int or not 0 < max_bytes <= 4 * 1024 * 1024):
+                raise ValueError("cloud_http_limits_invalid")
+            if self._connection is None:
+                # Default context validates certificates and hostname. This
+                # constructor and all TLS/HTTP IO execute in the owned worker.
+                self._connection = HTTPSConnection(*self._origin, timeout=timeout)
+            connection = self._connection
+            connection.timeout = timeout
+            reused = connection.sock is not None
+            if not reused:
+                # DNS/TCP/TLS finish before http.client can send the signed
+                # request. A connect timeout alone cannot identify which of
+                # those three sub-stages failed. Never infer queue settlement.
+                phase = "connect"
+                connection.connect()
+            if connection.sock is not None:
+                connection.sock.settimeout(timeout)
+            target = self._path + ("?" + parsed.query if parsed.query else "")
+            phase = "request"
+            connection.request("GET", target, headers=headers)
+            phase = "response_headers"
+            response = connection.getresponse()
+            with response:
+                # Do not follow redirects or retain a signed URL/error body.
+                if not 200 <= response.status < 300:
+                    raise HTTPError("", response.status, "cloud_http_status", None, BytesIO())
+                phase = "response_body"
+                payload = response.read(max_bytes + 1)
+                if len(payload) > max_bytes:
+                    raise ValueError("cloud_http_response_too_large")
+                if response.length not in (None, 0):
+                    raise IncompleteRead(b"", response.length)
+                if response.will_close:
+                    connection.close()
+                    self._connection = None
+            return payload
+        except BaseException as exc:
+            if phase != "validation":
+                _LOGGER.warning(
+                    "Cloud HTTPS request failed phase=%s reason=%s reused=%s elapsed_ms=%s; no automatic retry",
+                    phase, _failure_reason(exc), reused, round((monotonic() - started) * 1000),
+                )
+            self._close_locked()
+            if isinstance(exc, OSError) and not isinstance(exc, URLError):
+                raise URLError(exc) from exc
+            raise
+        finally:
+            self._lock.release()
+
+    def _close_locked(self):
+        self._closed = True
+        connection, self._connection = self._connection, None
+        if connection is not None:
+            connection.close()
+
+    def close(self) -> None:
+        """Owner-only cleanup AFTER draining its worker; never race request IO."""
+        if not self._lock.acquire(blocking=False):
+            raise RuntimeError("cloud_http_busy")
+        try:
+            self._close_locked()
+        finally:
+            self._lock.release()

custom_components/eybond_local/dessmonitor_cloud.py

@@ -399,18 +399,22 @@ def _decode_response_body(body: bytes) -> str:
     return text
 
 
-def _http_get_json(url: str, *, timeout: float) -> DessMonitorApiEnvelope:
+def _http_get_json(url: str, *, timeout: float, http_session=None) -> DessMonitorApiEnvelope:
     request = Request(
         url,
         headers={"User-Agent": "DessMonitor/3.6.2.1", "Accept": "application/json"},
         method="GET",
     )
     try:
-        with urlopen(request, timeout=_strict_timeout(timeout)) as response:
-            body = response.read(DEFAULT_MAX_RESPONSE_BYTES + 1)
-            if len(body) > DEFAULT_MAX_RESPONSE_BYTES:
-                raise DessMonitorCloudError("response_too_large")
-            payload = _decode_response_body(body)
+        if http_session is None:
+            with urlopen(request, timeout=_strict_timeout(timeout)) as response:
+                body = response.read(DEFAULT_MAX_RESPONSE_BYTES + 1)
+                if len(body) > DEFAULT_MAX_RESPONSE_BYTES:
+                    raise DessMonitorCloudError("response_too_large")
+        else:
+            body = http_session.get(url, headers=dict(request.header_items()), timeout=_strict_timeout(timeout),
+                                    max_bytes=DEFAULT_MAX_RESPONSE_BYTES)
+        payload = _decode_response_body(body)
     except HTTPError as exc:
         raise DessMonitorCloudError(f"http_error:{exc.code}") from exc
     except URLError as exc:
@@ -549,6 +553,7 @@ def fetch_signed_action(
     base_url: str = DEFAULT_BASE_URL,
     language: str = DEFAULT_LANGUAGE,
     timeout: float = DEFAULT_TIMEOUT,
+    http_session=None,
 ) -> DessMonitorApiEnvelope:
     stage = action.split("&", 2)[1].removeprefix("action=")
     try:
@@ -560,6 +565,7 @@ def fetch_signed_action(
                 language=language,
             ),
             timeout=timeout,
+            **({"http_session": http_session} if http_session is not None else {}),
         )
     except DessMonitorCloudError as exc:
         if exc.stage:
@@ -620,6 +626,7 @@ def send_device_control(
     base_url: str = DEFAULT_BASE_URL,
     language: str = DEFAULT_LANGUAGE,
     timeout: float = DEFAULT_TIMEOUT,
+    http_session=None,
 ) -> DessMonitorApiEnvelope:
     """Send one explicit DESSMonitor control action through a typed session."""
 
@@ -635,6 +642,7 @@ def send_device_control(
         base_url=base_url,
         language=language,
         timeout=timeout,
+        **({"http_session": http_session} if http_session is not None else {}),
     )
 
 

custom_components/eybond_local/smartess_cloud.py

@@ -213,7 +213,7 @@ def _decode_response_body(body: bytes) -> str:
     return text
 
 
-def _http_get_json(url: str, *, timeout: float = DEFAULT_TIMEOUT) -> ApiEnvelope:
+def _http_get_json(url: str, *, timeout: float = DEFAULT_TIMEOUT, http_session=None) -> ApiEnvelope:
     request = Request(
         url,
         headers={
@@ -223,8 +223,11 @@ def _http_get_json(url: str, *, timeout: float = DEFAULT_TIMEOUT) -> ApiEnvelope
         method="GET",
     )
     try:
-        with urlopen(request, timeout=timeout) as response:
-            payload = _decode_response_body(response.read())
+        if http_session is None:
+            with urlopen(request, timeout=timeout) as response:
+                payload = _decode_response_body(response.read())
+        else:
+            payload = _decode_response_body(http_session.get(url, headers=dict(request.header_items()), timeout=timeout))
     except HTTPError as exc:
         detail = exc.read().decode("utf-8", errors="replace").strip()
         raise SmartEssCloudError(f"http_error:{exc.code}:{detail}") from exc
@@ -380,6 +383,7 @@ def fetch_signed_action(
     app_id: str = DEFAULT_APP_ID,
     app_version: str = DEFAULT_APP_VERSION,
     timeout: float = DEFAULT_TIMEOUT,
+    http_session=None,
 ) -> ApiEnvelope:
     envelope = _http_get_json(
         build_signed_action_url(
@@ -391,6 +395,7 @@ def fetch_signed_action(
             app_version=app_version,
         ),
         timeout=timeout,
+        **({"http_session": http_session} if http_session is not None else {}),
     )
     if envelope.err != 0:
         raise SmartEssActionRejectedError(err=envelope.err, desc=envelope.desc)

custom_components/eybond_local/support/cloud_control_discovery.py

@@ -13,11 +13,14 @@
 
 from __future__ import annotations
 
+from contextlib import closing
 import logging
 from typing import Any
 
 from .. import valuecloud_cloud as valuecloud_cloud_module
+from ..cloud_http_session import CloudHttpSession
 from ..smartess_cloud import (
+    DEFAULT_BASE_URL,
     build_device_detail_action,
     build_device_settings_action,
     fetch_control_discovery_bundle_for_collector,
@@ -149,25 +152,29 @@ async def async_correlate(
 
         on_learning()
         progress(0.30, "testing")
-        result = await async_orchestrate_shadow_learning_settings(
-            settings_dat=settings_dat,
-            session=control_session,
-            pn=identity["pn"],
-            sn=identity["sn"],
-            devcode=identity["devcode"],
-            devaddr=identity["devaddr"],
-            dry_run=False,
-            confirm_cloud_write=True,
-            shadow_session_state="ready",
-            field_ids=[],
-            include_numeric=True,
-            all_choice_values=True,
-            max_fields=max_fields,
-            continue_on_error=True,
-            delay_seconds=0.0,
-            timeout=self.timeout_policy.action_request,
-            **dict(orchestrator_callbacks),
-        )
+        # The awaited dispatcher drains even repeatedly cancelled HTTP workers.
+        # Close only once that owner has returned; never race route recovery.
+        with closing(CloudHttpSession(DEFAULT_BASE_URL)) as http_session:
+            result = await async_orchestrate_shadow_learning_settings(
+                settings_dat=settings_dat,
+                session=control_session,
+                pn=identity["pn"],
+                sn=identity["sn"],
+                devcode=identity["devcode"],
+                devaddr=identity["devaddr"],
+                dry_run=False,
+                confirm_cloud_write=True,
+                shadow_session_state="ready",
+                field_ids=[],
+                include_numeric=True,
+                all_choice_values=True,
+                max_fields=max_fields,
+                continue_on_error=True,
+                delay_seconds=0.0,
+                timeout=self.timeout_policy.action_request,
+                http_session=http_session,
+                **dict(orchestrator_callbacks),
+            )
         result = dict(result)
         result["source"] = self.source_id
         result["metadata_only"] = False

custom_components/eybond_local/support/dessmonitor_active.py

@@ -2,8 +2,12 @@
 
 from __future__ import annotations
 
+from contextlib import closing
+
+from ..cloud_http_session import CloudHttpSession
 from ..collector_identity import pn_is_same_identity
 from ..dessmonitor_cloud import (
+    DEFAULT_BASE_URL,
     DessMonitorEvidenceBundle,
     fetch_read_only_evidence_for_session,
     login_with_password,
@@ -85,20 +89,24 @@ async def async_correlate(
         await start_shadow_route()
         on_learning()
         progress(0.30, "testing")
-        result = await async_orchestrate_dessmonitor_shadow_learning(
-            control_fields=bundle.control_fields,
-            session=cloud_session,
-            identity=bundle.identity,
-            confirm_cloud_write=True,
-            shadow_session_state="learning",
-            field_ids=(),
-            all_choice_values=True,
-            max_fields=max_fields,
-            continue_on_error=True,
-            delay_seconds=0.0,
-            timeout=self.timeout_policy.action_request,
-            **dict(orchestrator_callbacks),
-        )
+        # The awaited dispatcher drains even repeatedly cancelled HTTP workers.
+        # Close only once that owner has returned; never race route recovery.
+        with closing(CloudHttpSession(DEFAULT_BASE_URL)) as http_session:
+            result = await async_orchestrate_dessmonitor_shadow_learning(
+                control_fields=bundle.control_fields,
+                session=cloud_session,
+                identity=bundle.identity,
+                confirm_cloud_write=True,
+                shadow_session_state="learning",
+                field_ids=(),
+                all_choice_values=True,
+                max_fields=max_fields,
+                continue_on_error=True,
+                delay_seconds=0.0,
+                timeout=self.timeout_policy.action_request,
+                http_session=http_session,
+                **dict(orchestrator_callbacks),
+            )
         result = dict(result)
         result["source"] = self.source_id
         result["metadata_only"] = False

custom_components/eybond_local/support/shadow_learning/dessmonitor_orchestrator.py

@@ -293,6 +293,7 @@ async def async_orchestrate_dessmonitor_shadow_learning(
     timeout: float = DEFAULT_TIMEOUT,
     correlation_timeout_seconds: float = 2.0,
     action: Callable[..., DessMonitorApiEnvelope] = send_device_control,
+    http_session=None,
     on_progress: Callable[[int, int], None] | None = None,
 ) -> dict[str, Any]:
     """Correlate explicit DESSMonitor actions with writes on the exact route."""
@@ -401,6 +402,7 @@ async def async_orchestrate_dessmonitor_shadow_learning(
                 base_url=base_url,
                 language=language,
                 timeout=request_timeout,
+                **({"http_session": http_session} if http_session is not None else {}),
             )
             if type(envelope) is not DessMonitorApiEnvelope:
                 raise TypeError("dessmonitor_learning_response_invalid")

custom_components/eybond_local/support/shadow_learning/orchestrator.py

@@ -228,6 +228,7 @@ async def async_orchestrate_shadow_learning_settings(
     timeout: float = DEFAULT_TIMEOUT,
     correlation_timeout_seconds: float = 2.0,
     fetch_action: Callable[..., Any] = fetch_signed_action,
+    http_session=None,
     on_progress: Callable[[int, int], None] | None = None,
 ) -> dict[str, Any]:
     """Plan and execute one SmartESS learn-settings run with async per-attempt observation correlation.
@@ -315,6 +316,7 @@ async def async_orchestrate_shadow_learning_settings(
                 app_id=app_id,
                 app_version=app_version,
                 timeout=timeout,
+                **({"http_session": http_session} if http_session is not None else {}),
             )
             attempt["status"] = _CONTROL_STATUS_SENT
             attempt["delivery_outcome"] = "response"