From f5518f37057698f5ccc10d08c82b4aa5bc0d62e4 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Thu, 4 Jun 2026 15:28:20 -0400 Subject: [PATCH] sandbox/bridge: use orjson helper for batch key; bypass batcher for responses - Replace a stray function-local `import json` (json.dumps behind a bogus 'keeps json off the integration boot path' noqa) with HA's orjson json_bytes_sorted helper for the call-service batch key. - A response-returning entity call now bypasses the per-tick batcher. Coalescing forces every caller in a bucket to share one combined response, which is wrong when a caller needs its own value; response calls go out as their own single-entity RPC. The batcher is now fire-and-forget only, so its dead return_response plumbing is dropped. - Also removes development-phase references from this file's docstrings. Co-Authored-By: Claude Opus 4.8 (1M context) --- homeassistant/components/sandbox/bridge.py | 59 +++++++++++++--------- 1 file changed, 36 insertions(+), 23 deletions(-) diff --git a/homeassistant/components/sandbox/bridge.py b/homeassistant/components/sandbox/bridge.py index 7392c82bf94f..afb42c18f387 100644 --- a/homeassistant/components/sandbox/bridge.py +++ b/homeassistant/components/sandbox/bridge.py @@ -1,6 +1,6 @@ """Main-side bridge — owns the per-sandbox entity registry + outbound dispatch. -Responsibilities (Phase 5): +Responsibilities: * Hold a :class:`SandboxBridge` per sandbox group. Each one knows its :class:`Channel` plus the set of proxy entities the sandbox has @@ -24,9 +24,10 @@ Responsibilities (Phase 5): callers would have raised locally (``vol.Invalid`` → ``TypeError``, unknown service / entity → ``HomeAssistantError``). -Phase 8 adds the Store routing handlers (``sandbox/store_load`` / -``store_save`` / ``store_remove``). A per-group :class:`_SandboxStoreServer` -backs them, writing each key to ``/.storage/sandbox//``. +The Store routing handlers (``sandbox/store_load`` / +``store_save`` / ``store_remove``) are backed by a per-group +:class:`_SandboxStoreServer`, writing each key to +``/.storage/sandbox//``. Scope isolation is by construction — each bridge owns one channel for one group, so a sandbox can't reach another sandbox's files. """ @@ -167,7 +168,9 @@ class _CallServiceBatcher: batcher gathers everything that arrived this tick, fires one ``sandbox/call_service`` per (domain, service, kwargs-shape) bucket with a multi-entity ``target.entity_id`` list, and resolves all the - waiting futures with the same response. + waiting futures together. Only fire-and-forget calls are batched — a + call needing a response bypasses the batcher — so there is no per-entity + response to demultiplex. Kwargs are not hashable (they include nested dicts/lists), so the key is the JSON-canonical form of the kwargs dict. Only entities that @@ -179,7 +182,7 @@ class _CallServiceBatcher: def __init__(self, bridge: SandboxBridge) -> None: """Initialise the batcher with its owning bridge.""" self._bridge = bridge - self._buckets: dict[tuple[str, str, str], _BatchBucket] = {} + self._buckets: dict[tuple[str, str, bytes], _BatchBucket] = {} self._flush_handle: asyncio.Handle | None = None async def enqueue( @@ -190,14 +193,14 @@ class _CallServiceBatcher: sandbox_entity_id: str, service_data: dict[str, Any], context_id: str | None = None, - return_response: bool = False, ) -> Any: - """Queue one entity into the next batched ``call_service`` RPC.""" - import json # noqa: PLC0415 — local import keeps json off integration boot path + """Queue one entity into the next batched ``call_service`` RPC. - kwargs_key = json.dumps( - service_data, sort_keys=True, separators=(",", ":"), default=str - ) + Only fire-and-forget calls are batched; a call that needs a response + bypasses the batcher (see :meth:`SandboxBridge.async_call_service`). + """ + # Sorted canonical form keys the bucket; bytes are hashable. + kwargs_key = json_helper.json_bytes_sorted(service_data) bucket_key = (domain, service, kwargs_key) bucket = self._buckets.get(bucket_key) if bucket is None: @@ -207,7 +210,6 @@ class _CallServiceBatcher: service=service, service_data=service_data, context_id=context_id, - return_response=return_response, future=future, ) self._buckets[bucket_key] = bucket @@ -238,7 +240,7 @@ class _CallServiceBatcher: target={"entity_id": bucket.sandbox_entity_ids}, service_data=bucket.service_data, context_id=bucket.context_id, - return_response=bucket.return_response, + return_response=False, ) except BaseException as err: # noqa: BLE001 if not bucket.future.done(): @@ -256,7 +258,6 @@ class _BatchBucket: service: str service_data: dict[str, Any] context_id: str | None - return_response: bool future: asyncio.Future[Any] sandbox_entity_ids: list[str] = field(default_factory=list) @@ -325,24 +326,36 @@ class SandboxBridge: ) -> Any: """Forward one entity service call to the sandbox. - Calls made in the same tick with matching ``(domain, service, - service_data)`` coalesce into a single RPC with a multi-entity - target. + Fire-and-forget calls made in the same tick with matching ``(domain, + service, service_data)`` coalesce into a single RPC with a multi-entity + target. A call that needs a response (``return_response=True``) skips + the batcher: coalescing forces every caller in a bucket to share one + combined response, so a response call goes out as its own + single-entity RPC instead. ``context`` is the main-side Context driving the entity call. It is - remembered here (before the batcher reduces it to a bare id) so that + remembered here (before the id is reduced to a bare wire value) so that when the sandbox echoes the same id back on a resulting state change or event, :meth:`_resolve_context` restores the original ``parent_id`` / ``user_id`` instead of minting a fresh attribution. """ self._remember_context(context) + context_id = context.id if context is not None else None + if return_response: + return await self._raw_call_service( + domain=domain, + service=service, + target={"entity_id": [sandbox_entity_id]}, + service_data=service_data, + context_id=context_id, + return_response=True, + ) return await self._batcher.enqueue( domain=domain, service=service, sandbox_entity_id=sandbox_entity_id, service_data=service_data, - context_id=context.id if context is not None else None, - return_response=return_response, + context_id=context_id, ) async def _raw_call_service( @@ -590,7 +603,7 @@ class SandboxBridge: return pb.UnregisterServiceResult(ok=True, removed=True) async def _handle_store_load(self, msg: pb.StoreLoad) -> pb.StoreLoadResult: - """Serve a sandbox-side ``Store.async_load`` (Phase 8).""" + """Serve a sandbox-side ``Store.async_load``.""" data = await self._store_server.async_load(_validate_key(msg.key)) result = pb.StoreLoadResult() if data is not None: @@ -598,7 +611,7 @@ class SandboxBridge: return result async def _handle_store_save(self, msg: pb.StoreSave) -> pb.StoreSaveResult: - """Persist a sandbox-side ``Store.async_save`` flush (Phase 8).""" + """Persist a sandbox-side ``Store.async_save`` flush.""" await self._store_server.async_save( _validate_key(msg.key), struct_to_dict(msg.data) )