Reduce memory used by automation and script traces (#183260)

Co-authored-by: Claude <noreply@anthropic.com>

homeassistant/components/trace/models.py

@@ -1,7 +1,6 @@
 """Containers for a script or automation trace."""
 
 import abc
-from collections import deque
 from collections.abc import Iterator
 from dataclasses import dataclass
 import datetime as dt
@@ -81,7 +80,7 @@ class ActionTrace(BaseTrace):
         context: Context,
     ) -> None:
         """Container for script trace."""
-        self._trace: dict[str, deque[TraceElement]] | None = None
+        self._trace: dict[str, list[TraceElement]] | None = None
         self._config = config
         self._blueprint_inputs = blueprint_inputs
         self.context: Context = context
@@ -98,7 +97,7 @@ class ActionTrace(BaseTrace):
             trace_set_child_id(self.key, self.run_id)
         trace_id_set((self.key, self.run_id))
 
-    def set_trace(self, trace: dict[str, deque[TraceElement]] | None) -> None:
+    def set_trace(self, trace: dict[str, list[TraceElement]] | None) -> None:
         """Set action trace."""
         self._trace = trace
 
@@ -111,6 +110,10 @@ class ActionTrace(BaseTrace):
         self._timestamp_finish = dt_util.utcnow()
         self._state = "stopped"
         self._script_execution = script_execution_get()
+        if self._trace:
+            for elements in self._trace.values():
+                for element in elements:
+                    element.finish()
 
     @override
     def as_extended_dict(self) -> dict[str, Any]:

homeassistant/helpers/trace.py

@@ -1,10 +1,10 @@
 """Helpers for script and condition tracing."""
 
-from collections import deque
 from collections.abc import Callable, Coroutine, Generator
 from contextlib import contextmanager
 from contextvars import ContextVar
 from functools import wraps
+import sys
 from typing import Any, Literal, overload, override
 
 from homeassistant.core import ServiceResponse
@@ -40,7 +40,7 @@ class TraceElement:
         self.reuse_by_child = False
         self._timestamp = dt_util.utcnow()
 
-        self._last_variables = variables_cv.get() or {}
+        self._last_variables: dict[str, Any] | None = variables_cv.get() or {}
         self.update_variables(variables)
 
     @override
@@ -87,7 +87,7 @@ class TraceElement:
         """Update variables."""
         if variables is None:
             variables = {}
-        last_variables = self._last_variables
+        last_variables = self._last_variables or {}
         # variables is often a ChainMap which is costly to iterate, so flatten
         # it once and reuse the snapshot for both the baseline and the diff.
         snapshot = dict(variables)
@@ -98,6 +98,12 @@ class TraceElement:
             if key not in last_variables or last_variables[key] != value
         }
 
+    def finish(self) -> None:
+        """Release data which is only needed while the trace is recorded."""
+        # The previous step's full variables snapshot is only needed to
+        # compute the changed variables of this step.
+        self._last_variables = None
+
     def as_dict(self) -> dict[str, Any]:
         """Return dictionary version of this TraceElement."""
         result: dict[str, Any] = {"path": self.path, "timestamp": self._timestamp}
@@ -121,7 +127,7 @@ class TraceElement:
 
 # Context variables for tracing
 # Current trace
-trace_cv: ContextVar[dict[str, deque[TraceElement]] | None] = ContextVar(
+trace_cv: ContextVar[dict[str, list[TraceElement]] | None] = ContextVar(
     "trace_cv", default=None
 )
 # Stack of TraceElements
@@ -218,7 +224,8 @@ def trace_path_get() -> str:
     """Return a string representing the current location in the config tree."""
     if not (path := trace_path_stack_cv.get()):
         return ""
-    return "/".join(path)
+    # The same paths are recorded on every run, share a single string
+    return sys.intern("/".join(path))
 
 
 def trace_append_element(
@@ -230,19 +237,23 @@ def trace_append_element(
         trace = {}
         trace_cv.set(trace)
     if (path := trace_element.path) not in trace:
-        trace[path] = deque(maxlen=maxlen)
-    trace[path].append(trace_element)
+        trace[path] = [trace_element]
+        return
+    elements = trace[path]
+    if maxlen is not None and len(elements) >= maxlen:
+        del elements[0]
+    elements.append(trace_element)
 
 
 @overload
-def trace_get(clear: Literal[True] = True) -> dict[str, deque[TraceElement]]: ...
+def trace_get(clear: Literal[True] = True) -> dict[str, list[TraceElement]]: ...
 
 
 @overload
-def trace_get(clear: Literal[False]) -> dict[str, deque[TraceElement]] | None: ...
+def trace_get(clear: Literal[False]) -> dict[str, list[TraceElement]] | None: ...
 
 
-def trace_get(clear: bool = True) -> dict[str, deque[TraceElement]] | None:
+def trace_get(clear: bool = True) -> dict[str, list[TraceElement]] | None:
     """Return the current trace.
 
     When clear is True the trace is reset and a fresh (empty) trace is

tests/components/trace/test_websocket_api.py

@@ -1,7 +1,7 @@
 """Test Trace websocket API."""
 
 import asyncio
-from collections import defaultdict, deque
+from collections import defaultdict
 import json
 from typing import Any
 from unittest.mock import patch
@@ -1677,7 +1677,7 @@ def _serialize_trace(not_triggered: bool, reason: str) -> dict[str, Any]:
     trace.not_triggered = not_triggered
     element = TraceElement({"trigger": {"idx": "0"}}, "trigger/0")
     element.set_result(reason=reason)
-    trace.set_trace({"trigger/0": deque([element])})
+    trace.set_trace({"trigger/0": [element]})
     trace.finished()
     return json.loads(json.dumps(trace.as_dict(), cls=ExtendedJSONEncoder))