From e83f2c1adcda3c7d492fa402e7ba59f1c0b36286 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sun, 4 Oct 2026 11:01:06 +0100 Subject: [PATCH] perf(rules): check custom-rule module state with one shared snapshot per call After every custom-rule call the host re-fingerprinted the module state of every loaded rule code, walking all reachable globals once per code. Each exact fingerprint walk now also traces what it touched: the identity and type of every object it visited, the classes it descended into, and the empty instances whose immutability it checked. One shared snapshot then replays the union of those walks over every code still being monitored and is compared with the snapshot taken when their fingerprints last matched. The snapshot descends only where a traced walk descended and treats any untraced object, any error, or more than the node bound per walk as a reason to fall back, so it never touches or runs code on an object the per-code walks did not. An equal snapshot proves no fingerprint changed; otherwise every remaining code is fingerprinted exactly as before and the trace is rebuilt. Codes sharing a module are fingerprinted once, and codes already stateful are no longer checked. Both walkers share one structural dispatch so they always take the same branch. Classification is unchanged: equality tests compare the monitor with full re-fingerprinting after every call over a corpus of stateful and stateless rule modules, raising and side-effecting attribute hooks the fingerprints never reach, and seeded random mutations, and a CLI test checks split hosts report what one host reports. --- fensu.toml | 6 + .../_helpers/host/custom_evaluation.py | 38 +- .../rule_engine/_helpers/host/module_state.py | 467 ++++++++++++++++-- .../classes/module_state_monitor.py | 100 ++++ src/sqlbuild/rule_engine/models.py | 17 + .../cli/commands/main/rules/_test_types.py | 11 + .../cli/commands/main/rules/helpers.py | 49 ++ .../rules/test_custom_rule_hermeticity.py | 73 +++ .../rule_engine/_helpers/host/_test_types.py | 30 ++ .../rule_engine/_helpers/host/helpers.py | 293 +++++++++++ .../_helpers/host/test_module_state.py | 427 ++++++++++++++++ 11 files changed, 1432 insertions(+), 79 deletions(-) create mode 100644 src/sqlbuild/rule_engine/classes/module_state_monitor.py create mode 100644 tests/unit/src/sqlbuild/rule_engine/_helpers/host/test_module_state.py diff --git a/fensu.toml b/fensu.toml index 8394307b0a..e855b92b7e 100644 --- a/fensu.toml +++ b/fensu.toml @@ -105,6 +105,12 @@ path = "src/sqlbuild/executor/migrations/_helpers/columns.py" symbols = ["apply_model_column_renames"] reason = "Each in-place column rename must re-read live columns immediately before its ALTER so overlapping builds reconcile instead of failing." +[[targets.python.rule_exceptions]] +rule = "FFH005" +path = "src/sqlbuild/rule_engine/classes/module_state_monitor.py" +symbols = ["ModuleStateMonitor._snapshot"] +reason = "A failed shared snapshot only disables the shortcut; exact per-rule fingerprints then run and keep their own failure behavior, so no snapshot error may fail a Rules run." + [[targets.python.rule_exceptions]] rule = "FFS120" path = "src/sqlbuild/integrations/dbt/_helpers/profile/render.py" diff --git a/src/sqlbuild/rule_engine/_helpers/host/custom_evaluation.py b/src/sqlbuild/rule_engine/_helpers/host/custom_evaluation.py index 832e467da7..9e9c501d90 100644 --- a/src/sqlbuild/rule_engine/_helpers/host/custom_evaluation.py +++ b/src/sqlbuild/rule_engine/_helpers/host/custom_evaluation.py @@ -10,12 +10,10 @@ from pathlib import Path from sqlbuild.compiler.compile.models import CompiledModel, CompiledProject -from sqlbuild.rule_engine._helpers.host.module_state import ( - module_state_token, - rule_namespaces, -) +from sqlbuild.rule_engine._helpers.host.module_state import rule_namespaces from sqlbuild.rule_engine._helpers.host.tracked_facts import tracked_fact_views from sqlbuild.rule_engine.classes.fact_reads import FactReads +from sqlbuild.rule_engine.classes.module_state_monitor import ModuleStateMonitor from sqlbuild.rule_engine.classes.project_tree import public_model from sqlbuild.rule_engine.classes.rule_context import ( EvaluationRuleContext, @@ -30,7 +28,6 @@ from sqlbuild.rule_engine.exceptions import ( HostCancelledError, NonHermeticRuleError, - OpaqueModuleStateError, RulesError, ) from sqlbuild.rule_engine.models import ( @@ -150,15 +147,14 @@ def __init__( self.untracked_codes: set[str] = set() self.uncacheable_codes: set[str] = set() self.stateful_codes: set[str] = set() - self._rules_root: Path = rules_root - self._namespaces: dict[str, tuple[dict[str, object], ...]] = ( - {rule.code: rule_namespaces((rule.check,)) for rule in rules} + self._state: ModuleStateMonitor | None = ( + ModuleStateMonitor( + namespaces={rule.code: rule_namespaces((rule.check,)) for rule in rules}, + rules_root=rules_root, + ) if reads is not None or detect_state - else {} + else None ) - self._states: dict[str, str | None] = { - code: self._state_token(namespaces) for code, namespaces in self._namespaces.items() - } def start(self) -> None: if self.reads is not None: @@ -176,22 +172,12 @@ def finish(self, *, code: str) -> frozenset[tuple[object, ...]] | None: self.untracked_codes.add(code) if reads.uncacheable: self.uncacheable_codes.add(code) - for rule_code, namespaces in self._namespaces.items(): - if rule_code in self.stateful_codes: - continue - state: str | None = self._state_token(namespaces) - if state is None or state != self._states[rule_code]: - self.stateful_codes.add(rule_code) - if reads is not None: - self.untracked_codes.add(rule_code) + changed: tuple[str, ...] = () if self._state is None else self._state.changed_codes() + self.stateful_codes.update(changed) + if reads is not None: + self.untracked_codes.update(changed) return None if observed is None else frozenset(observed) - def _state_token(self, namespaces: tuple[dict[str, object], ...]) -> str | None: - try: - return module_state_token(namespaces=namespaces, rules_root=self._rules_root) - except OpaqueModuleStateError: - return None - def _evaluate_pass( *, diff --git a/src/sqlbuild/rule_engine/_helpers/host/module_state.py b/src/sqlbuild/rule_engine/_helpers/host/module_state.py index e0d0a26a39..eba9b96c5c 100644 --- a/src/sqlbuild/rule_engine/_helpers/host/module_state.py +++ b/src/sqlbuild/rule_engine/_helpers/host/module_state.py @@ -6,15 +6,18 @@ import hashlib import os import re -from collections import deque -from collections.abc import Iterable +from abc import ABC, abstractmethod +from collections import OrderedDict, defaultdict, deque +from collections.abc import Callable, Collection, Iterable, Iterator, Mapping from decimal import Decimal from fractions import Fraction +from operator import itemgetter from pathlib import Path, PurePath -from types import BuiltinFunctionType, FunctionType, MethodType, ModuleType +from types import BuiltinFunctionType, CellType, FunctionType, MethodType, ModuleType from typing import Any, cast from sqlbuild.rule_engine.exceptions import OpaqueModuleStateError +from sqlbuild.rule_engine.models import ModuleStateFingerprint, ModuleStateTrace _MAX_STATE_NODES: int = 50_000 _ATOMIC_TYPES: tuple[type, ...] = ( @@ -35,6 +38,35 @@ ) _IMMUTABLE_LIBRARY_MODULES: frozenset[str] = frozenset({"typing", "types", "builtins"}) _SKIPPED_FUNCTION_ATTRIBUTES: frozenset[str] = frozenset({"__sqlbuild_rule__", "__wrapped__"}) +_PLAIN_TYPES: frozenset[type] = frozenset({str, int, bytes}) +_BUILTIN_ATOM_TYPES: frozenset[type] = frozenset( + {type(None), bool, float, complex, bytearray, range, re.Pattern, Decimal} +) +_UNTRACED_TYPES: frozenset[type] = _PLAIN_TYPES | _BUILTIN_ATOM_TYPES +_BUILTIN_SEQUENCES: frozenset[type] = frozenset({tuple, list, deque}) +_BUILTIN_MAPPINGS: frozenset[type] = frozenset({dict, defaultdict, OrderedDict}) +_BUILTIN_SETS: frozenset[type] = frozenset({set, frozenset}) +_BUILTIN_CONTAINERS: frozenset[type] = ( + _BUILTIN_SEQUENCES + | _BUILTIN_MAPPINGS + | _BUILTIN_SETS + | {FunctionType, MethodType, BuiltinFunctionType, ModuleType, type} +) +_NAMESPACE: str = "namespace" +_REFERENCE: str = "ref" +_OBJECT: str = "object" +_ATOM: str = "atom" +_SET: str = "set" +_ITEMS: str = "items" +_PLAIN_ITEMS: str = "plain-items" +_MAPPING: str = "mapping" +_CELLS: str = "cells" +_ATTRIBUTES: str = "attributes" +_EMPTY_INSTANCE: str = "empty-instance" +_OWNED_MODULE: tuple[str] = ("owned-module",) +_OWNED_FUNCTION: tuple[str] = ("owned-function",) +_EMPTY_CELL: tuple[str] = ("empty-cell",) +_DESCENDED: tuple[str] = ("descended",) def rule_namespaces(rules: Iterable[object]) -> tuple[dict[str, object], ...]: @@ -51,18 +83,125 @@ def rule_namespaces(rules: Iterable[object]) -> tuple[dict[str, object], ...]: def module_state_token(*, namespaces: tuple[dict[str, object], ...], rules_root: Path) -> str: """Fingerprint every mutable value reachable from rule-module globals.""" - walker: _StateWalker = _StateWalker(rules_root=f"{rules_root.resolve()}{os.sep}") + return module_state_fingerprint( + namespaces=namespaces, rules_root=rules_root_prefix(rules_root) + ).token + + +def rules_root_prefix(rules_root: Path) -> str: + """Return the resolved path prefix that marks code and modules as owned by the project.""" + + return f"{rules_root.resolve()}{os.sep}" + + +def module_state_fingerprint( + *, namespaces: tuple[dict[str, object], ...], rules_root: str +) -> ModuleStateFingerprint: + """Fingerprint rule-module state exactly and trace what the walk touched and decided.""" + + walker: _StateWalker = _StateWalker(rules_root=rules_root) + for namespace in namespaces: + walker.visit_namespace(namespace) + return ModuleStateFingerprint( + token=walker.digest.hexdigest(), + trace=ModuleStateTrace( + touched=walker.touched, + descended_types=frozenset(walker.descended_types), + checked_instances=frozenset(walker.checked_instances), + ), + ) + + +def module_state_snapshot( + *, + namespaces: Iterable[dict[str, object]], + rules_root: str, + trace: ModuleStateTrace, + walks: int, +) -> list[object]: + """Replay `walks` traced fingerprint walks at once and record what each would observe.""" + + walker: _SnapshotWalker = _SnapshotWalker( + rules_root=rules_root, trace=trace, max_nodes=_MAX_STATE_NODES * walks + ) for namespace in namespaces: walker.visit_namespace(namespace) - return walker.digest.hexdigest() + return walker.observed -class _StateWalker: +class _Walker(ABC): + """Dispatch one reachable object to the branch every module-state walk takes for it.""" + def __init__(self, *, rules_root: str) -> None: - self.digest: Any = hashlib.sha256() self._rules_root: str = rules_root + + @abstractmethod + def visit(self, value: object) -> None: ... + + def _visit_structure(self, value: object) -> None: + if isinstance(value, ModuleType): + location: object = getattr(value, "__file__", None) + self._visit_module( + module=value, + owned=isinstance(location, str) and location.startswith(self._rules_root), + ) + elif isinstance(value, dict): + self._visit_mapping(cast(dict[object, object], value)) + elif isinstance(value, (list, tuple, deque)): + self._visit_items(value) + elif isinstance(value, (set, frozenset)): + self._visit_set(value) + elif isinstance(value, FunctionType): + self._visit_function(value) + elif isinstance(value, MethodType): + self.visit(value.__func__) + self.visit(value.__self__) + elif isinstance(value, type): + self._visit_type(value) + elif hasattr(value, "cache_info") and hasattr(value, "__wrapped__"): + self._visit_cache(value) + elif isinstance(value, BuiltinFunctionType): + self._visit_builtin(value) + else: + self._visit_instance(value) + + @abstractmethod + def _visit_module(self, *, module: ModuleType, owned: bool) -> None: ... + + @abstractmethod + def _visit_mapping(self, value: dict[object, object]) -> None: ... + + @abstractmethod + def _visit_items(self, items: Collection[object]) -> None: ... + + @abstractmethod + def _visit_set(self, value: Collection[object]) -> None: ... + + @abstractmethod + def _visit_function(self, value: FunctionType) -> None: ... + + @abstractmethod + def _visit_type(self, value: type) -> None: ... + + @abstractmethod + def _visit_cache(self, value: Any) -> None: ... + + @abstractmethod + def _visit_builtin(self, value: BuiltinFunctionType) -> None: ... + + @abstractmethod + def _visit_instance(self, value: object) -> None: ... + + +class _StateWalker(_Walker): + def __init__(self, *, rules_root: str) -> None: + self.digest: Any = hashlib.sha256() + self.module_names: set[str] = set() + self.touched: dict[int, int] = {} + self.descended_types: set[int] = set() + self.checked_instances: set[int] = set() + super().__init__(rules_root=rules_root) self._namespaces: set[int] = set() - self._module_names: set[str] = set() self._visited: dict[int, int] = {} self._nodes: int = 0 @@ -71,7 +210,7 @@ def visit_namespace(self, namespace: dict[str, object]) -> None: return self._namespaces.add(id(namespace)) name: object = namespace.get("__name__") - self._module_names.add(str(name)) + self.module_names.add(str(name)) self.emit(f"namespace:{name}") for key, value in sorted(namespace.items(), key=lambda item: item[0]): if key.startswith("__") and key.endswith("__"): @@ -82,6 +221,11 @@ def visit_namespace(self, namespace: dict[str, object]) -> None: def _owned_code(self, function: FunctionType) -> bool: return function.__code__.co_filename.startswith(self._rules_root) + def _trace(self, value: object) -> None: + kind: type = type(value) + if kind not in _UNTRACED_TYPES: + self.touched[id(value)] = id(kind) + def emit(self, text: str) -> None: self.digest.update(text.encode()) self.digest.update(b"\0") @@ -90,6 +234,7 @@ def visit(self, value: object) -> None: self._nodes += 1 if self._nodes > _MAX_STATE_NODES: raise OpaqueModuleStateError("custom rule module state is too large to observe") + self._trace(value) if isinstance(value, _ATOMIC_TYPES) and not isinstance(value, (list, dict, set)): self.emit(f"atom:{type(value).__qualname__}:{value!r}") return @@ -102,39 +247,29 @@ def visit(self, value: object) -> None: self.emit(f"object:{identity}:{type(value).__qualname__}") self._visit_structure(value) - def _visit_structure(self, value: object) -> None: - if isinstance(value, ModuleType): - location: object = getattr(value, "__file__", None) - if isinstance(location, str) and location.startswith(self._rules_root): - self.visit_namespace(vars(value)) - else: - self.emit(f"module:{value.__name__}") - elif isinstance(value, dict): - self._visit_mapping(cast(dict[object, object], value)) - elif isinstance(value, (list, tuple, deque)): - self.emit(f"len:{len(value)}") - for item in value: - self.visit(item) - elif isinstance(value, (set, frozenset)): - self.emit(f"len:{len(value)}") - tokens: list[str] = sorted(_atomic_text(item) for item in value) - for token in tokens: - self.emit(token) - elif isinstance(value, FunctionType): - self._visit_function(value) - elif isinstance(value, MethodType): - self.visit(value.__func__) - self.visit(value.__self__) - elif isinstance(value, type): - self._visit_type(value) - elif hasattr(value, "cache_info") and hasattr(value, "__wrapped__"): - cached: Any = cast(Any, value) - self.emit(f"cache:{cached.cache_info().currsize}") - self.visit(cached.__wrapped__) - elif isinstance(value, BuiltinFunctionType): - self.emit(f"builtin:{getattr(value, '__module__', '')}:{value.__qualname__}") + def _visit_module(self, *, module: ModuleType, owned: bool) -> None: + if owned: + self.visit_namespace(vars(module)) else: - self._visit_instance(value) + self.emit(f"module:{module.__name__}") + + def _visit_items(self, items: Collection[object]) -> None: + self.emit(f"len:{len(items)}") + for item in items: + self.visit(item) + + def _visit_set(self, value: Collection[object]) -> None: + self.emit(f"len:{len(value)}") + tokens: list[str] = sorted(_atomic_text(value=item, observe=self._trace) for item in value) + for token in tokens: + self.emit(token) + + def _visit_cache(self, value: Any) -> None: + self.emit(f"cache:{value.cache_info().currsize}") + self.visit(value.__wrapped__) + + def _visit_builtin(self, value: BuiltinFunctionType) -> None: + self.emit(f"builtin:{getattr(value, '__module__', '')}:{value.__qualname__}") def _visit_mapping(self, value: dict[object, object]) -> None: self.emit(f"len:{len(value)}") @@ -166,13 +301,14 @@ def _visit_function(self, value: FunctionType) -> None: self.visit(contents) def _visit_type(self, value: type) -> None: - owned: bool = value.__module__ in self._module_names or any( + owned: bool = value.__module__ in self.module_names or any( isinstance(item, FunctionType) and self._owned_code(item) for item in vars(value).values() ) if not owned: self.emit(f"type:{value.__module__}:{value.__qualname__}") return + self.descended_types.add(id(value)) for name, item in sorted(vars(value).items(), key=lambda entry: entry[0]): if name.startswith("__") and name.endswith("__"): continue @@ -180,15 +316,9 @@ def _visit_type(self, value: type) -> None: self.visit(item) def _visit_instance(self, value: object) -> None: - state: dict[str, object] = {} - attributes: object = getattr(value, "__dict__", None) - if isinstance(attributes, dict): - state.update(attributes) - for owner in type(value).__mro__: - for slot in getattr(owner, "__slots__", ()): - if isinstance(slot, str) and hasattr(value, slot): - state[slot] = getattr(value, slot) - if not state and type(value).__module__ not in self._module_names: + state: dict[str, object] = _instance_state(value) + if not state and type(value).__module__ not in self.module_names: + self.checked_instances.add(id(value)) if _is_immutable_library_value(value): self.emit(f"value:{type(value).__module__}:{type(value).__qualname__}") return @@ -200,11 +330,242 @@ def _visit_instance(self, value: object) -> None: self.visit(item) -def _atomic_text(value: object) -> str: +class _SnapshotWalker(_Walker): + """Replay traced `_StateWalker` walks, recording what they observe without hashing it.""" + + def __init__(self, *, rules_root: str, trace: ModuleStateTrace, max_nodes: int) -> None: + super().__init__(rules_root=rules_root) + self.observed: list[object] = [] + self._trace: ModuleStateTrace = trace + self._touched: dict[int, int] = trace.touched + self._max_nodes: int = max_nodes + self._nodes: int = 0 + self._namespaces: set[int] = set() + self._visited: dict[int, int] = {} + + def visit_namespace(self, namespace: dict[str, object]) -> None: + identity: int = id(namespace) + if identity in self._namespaces: + return + self._namespaces.add(identity) + items: list[tuple[str, object]] = sorted(_named_items(namespace), key=itemgetter(0)) + self.observed.append( + ( + _NAMESPACE, + identity, + f"namespace:{namespace.get('__name__')}", + [key for key, _ in items], + ) + ) + for _, value in items: + self.visit(value) + + def visit(self, value: object) -> None: + self._nodes += 1 + if self._nodes > self._max_nodes: + raise OpaqueModuleStateError("custom rule module state is too large to snapshot") + kind: type = type(value) + if kind in _PLAIN_TYPES: + self.observed.append(value) + return + if kind in _BUILTIN_ATOM_TYPES: + self.observed.append(_atom(value)) + return + if self._touched.get(id(value)) != id(kind): + raise OpaqueModuleStateError("custom rule module state reached an untraced value") + if ( + kind not in _BUILTIN_CONTAINERS + and isinstance(value, _ATOMIC_TYPES) + and not isinstance(value, (list, dict, set)) + ): + self.observed.append(_atom(value)) + return + identity: int = id(value) + seen: int | None = self._visited.get(identity) + if seen is not None: + self.observed.append((_REFERENCE, seen)) + return + self._visited[identity] = len(self._visited) + self.observed.append((_OBJECT, identity, kind.__qualname__)) + if kind is tuple or kind is list: + self._visit_items(cast(Collection[object], value)) + elif kind is dict: + self._visit_mapping(cast(dict[object, object], value)) + elif kind is FunctionType: + self._visit_function(cast(FunctionType, value)) + else: + self._visit_structure(value) + + def _count(self, nodes: int) -> None: + self._nodes += nodes + if self._nodes > self._max_nodes: + raise OpaqueModuleStateError("custom rule module state is too large to snapshot") + + def _require_traced(self, value: object) -> None: + if self._trace.touched.get(id(value)) != id(type(value)): + raise OpaqueModuleStateError("custom rule module state reached an untraced value") + + def _require_member_traced(self, value: object) -> None: + if type(value) not in _UNTRACED_TYPES: + self._require_traced(value) + + def _visit_module(self, *, module: ModuleType, owned: bool) -> None: + if owned: + self.observed.append(_OWNED_MODULE) + self.visit_namespace(vars(module)) + else: + self.observed.append(f"module:{module.__name__}") + + def _visit_cache(self, value: Any) -> None: + self.observed.append(f"cache:{value.cache_info().currsize}") + self.visit(value.__wrapped__) + + def _visit_builtin(self, value: BuiltinFunctionType) -> None: + self.observed.append(f"builtin:{getattr(value, '__module__', '')}:{value.__qualname__}") + + def _visit_items(self, items: Collection[object]) -> None: + if type(items) in _BUILTIN_SEQUENCES and _plain(items): + self._count(len(items)) + self.observed.append((_PLAIN_ITEMS, list(items))) + return + self.observed.append((_ITEMS, len(items))) + for item in items: + self.visit(item) + + def _visit_mapping(self, value: dict[object, object]) -> None: + size: int = len(value) + factory: object = getattr(value, "default_factory", None) + self.observed.append((_MAPPING, factory is None)) + if factory is not None: + self.visit(factory) + if type(value) in _BUILTIN_MAPPINGS and _plain(value.keys()) and _plain(value.values()): + self._count(2 * size) + self.observed.append(list(value.items())) + return + self.observed.append(size) + for key, item in value.items(): + self.visit(key) + self.visit(item) + + def _visit_set(self, value: Collection[object]) -> None: + size: int = len(value) + self._count(size) + if type(value) in _BUILTIN_SETS and _plain(value): + self.observed.append((_SET, set(value))) + return + self.observed.append( + ( + _SET, + size, + sorted( + _atomic_text(value=item, observe=self._require_member_traced) for item in value + ), + ) + ) + + def _visit_function(self, value: FunctionType) -> None: + if not value.__code__.co_filename.startswith(self._rules_root): + self.observed.append(f"function:{value.__module__}:{value.__qualname__}") + return + self.observed.append(_OWNED_FUNCTION) + self.visit_namespace(value.__globals__) + self.visit(value.__defaults__) + self.visit(value.__kwdefaults__) + self._visit_attributes( + (name, item) + for name, item in vars(value).items() + if name not in _SKIPPED_FUNCTION_ATTRIBUTES + ) + cells: tuple[CellType, ...] = value.__closure__ or () + self.observed.append((_CELLS, len(cells))) + for cell in cells: + try: + contents: object = cell.cell_contents + except ValueError: + self.observed.append(_EMPTY_CELL) + continue + self.visit(contents) + + def _visit_type(self, value: type) -> None: + owns_code: bool = any( + type(item) is FunctionType + and cast(FunctionType, item).__code__.co_filename.startswith(self._rules_root) + for item in vars(value).values() + ) + self.observed.append(f"type:{value.__module__}:{value.__qualname__}") + self.observed.append(owns_code) + if id(value) in self._trace.descended_types: + self.observed.append(_DESCENDED) + self._visit_attributes(_named_items(vars(value))) + + def _visit_instance(self, value: object) -> None: + state: dict[str, object] = _instance_state(value) + if not state: + checked: bool = id(value) in self._trace.checked_instances + self.observed.append( + ( + _EMPTY_INSTANCE, + f"{type(value).__module__}", + checked and _is_immutable_library_value(value), + checked, + ) + ) + return + self._visit_attributes(state.items()) + + def _visit_attributes(self, attributes: Iterable[tuple[str, object]]) -> None: + listed: list[tuple[str, object]] = list(attributes) + if not _plain_names(name for name, _ in listed): + raise OpaqueModuleStateError("custom rule module state has a non-string attribute") + items: list[tuple[str, object]] = sorted(listed, key=itemgetter(0)) + self.observed.append((_ATTRIBUTES, [name for name, _ in items])) + for _, item in items: + self.visit(item) + + +def _plain(values: Iterable[object]) -> bool: + return {*map(type, values)} <= _PLAIN_TYPES + + +def _plain_names(names: Iterable[object]) -> bool: + return {*map(type, names)} <= {str} + + +def _atom(value: object) -> tuple[str, ...]: + if type(value) is re.Pattern: + pattern: re.Pattern[str] = cast(re.Pattern[str], value) + return (_ATOM, "Pattern", pattern.pattern, str(pattern.flags)) + if type(value) is bool or value is None: + return (_ATOM, repr(value)) + return (_ATOM, type(value).__qualname__, repr(value)) + + +def _named_items(namespace: Mapping[str, object]) -> Iterator[tuple[str, object]]: + for key, value in namespace.items(): + if type(key) is not str: + raise OpaqueModuleStateError("custom rule module state has a non-string name") + if not (key.startswith("__") and key.endswith("__")): + yield key, value + + +def _instance_state(value: object) -> dict[str, object]: + state: dict[str, object] = {} + attributes: object = getattr(value, "__dict__", None) + if isinstance(attributes, dict): + state.update(cast(dict[str, object], attributes)) + for owner in type(value).__mro__: + for slot in getattr(owner, "__slots__", ()): + if isinstance(slot, str) and hasattr(value, slot): + state[slot] = getattr(value, slot) + return state + + +def _atomic_text(*, value: object, observe: Callable[[object], None]) -> str: + observe(value) if isinstance(value, _ATOMIC_TYPES): return f"{type(value).__qualname__}:{value!r}" if isinstance(value, tuple): - return "(" + ",".join(_atomic_text(item) for item in value) + ")" + return "(" + ",".join(_atomic_text(value=item, observe=observe) for item in value) + ")" raise OpaqueModuleStateError( f"custom rule module state holds an unhashable-order {type(value).__qualname__}" ) diff --git a/src/sqlbuild/rule_engine/classes/module_state_monitor.py b/src/sqlbuild/rule_engine/classes/module_state_monitor.py new file mode 100644 index 0000000000..98435f2506 --- /dev/null +++ b/src/sqlbuild/rule_engine/classes/module_state_monitor.py @@ -0,0 +1,100 @@ +"""Exact per-rule module-state change detection with one shared snapshot per check.""" + +from __future__ import annotations + +from itertools import chain +from pathlib import Path + +from sqlbuild.rule_engine._helpers.host.module_state import ( + module_state_fingerprint, + module_state_snapshot, + rules_root_prefix, +) +from sqlbuild.rule_engine.exceptions import OpaqueModuleStateError +from sqlbuild.rule_engine.models import ModuleStateFingerprint, ModuleStateTrace + +_EMPTY_TRACE: ModuleStateTrace = ModuleStateTrace( + touched={}, descended_types=frozenset(), checked_instances=frozenset() +) + + +class ModuleStateMonitor: + """Report rule codes whose module-state fingerprint differs from when monitoring started.""" + + def __init__( + self, *, namespaces: dict[str, tuple[dict[str, object], ...]], rules_root: Path + ) -> None: + self._rules_root: str = rules_root_prefix(rules_root) + self._namespaces: dict[str, tuple[dict[str, object], ...]] = namespaces + self._trace: ModuleStateTrace = _EMPTY_TRACE + states: dict[str, ModuleStateFingerprint | None] = self._fingerprints(tuple(namespaces)) + self._initial: dict[str, str] = { + code: state.token for code, state in states.items() if state is not None + } + self._unobservable: tuple[str, ...] = tuple( + code for code, state in states.items() if state is None + ) + self._baseline: list[object] | None = self._verified_snapshot(states) + + def changed_codes(self) -> tuple[str, ...]: + """Return codes whose module state changed since monitoring started, each only once.""" + + changed: list[str] = list(self._unobservable) + self._unobservable = () + if not self._initial: + return tuple(changed) + if self._baseline is not None and self._snapshot() == self._baseline: + return tuple(changed) + states: dict[str, ModuleStateFingerprint | None] = self._fingerprints(tuple(self._initial)) + for code, state in states.items(): + if state is None or state.token != self._initial[code]: + changed.append(code) + del self._initial[code] + self._baseline = self._verified_snapshot(states) + return tuple(changed) + + def _fingerprints(self, codes: tuple[str, ...]) -> dict[str, ModuleStateFingerprint | None]: + by_namespaces: dict[tuple[int, ...], ModuleStateFingerprint | None] = {} + states: dict[str, ModuleStateFingerprint | None] = {} + for code in codes: + namespaces: tuple[dict[str, object], ...] = self._namespaces[code] + key: tuple[int, ...] = tuple(map(id, namespaces)) + if key not in by_namespaces: + try: + by_namespaces[key] = module_state_fingerprint( + namespaces=namespaces, rules_root=self._rules_root + ) + except OpaqueModuleStateError: + by_namespaces[key] = None + states[code] = by_namespaces[key] + return states + + def _verified_snapshot( + self, states: dict[str, ModuleStateFingerprint | None] + ) -> list[object] | None: + traces: tuple[ModuleStateTrace, ...] = tuple( + state.trace + for code, state in states.items() + if state is not None and code in self._initial + ) + self._trace = ModuleStateTrace( + touched=dict(chain.from_iterable(trace.touched.items() for trace in traces)), + descended_types=frozenset( + chain.from_iterable(trace.descended_types for trace in traces) + ), + checked_instances=frozenset( + chain.from_iterable(trace.checked_instances for trace in traces) + ), + ) + return self._snapshot() + + def _snapshot(self) -> list[object] | None: + try: + return module_state_snapshot( + namespaces=chain.from_iterable(map(self._namespaces.__getitem__, self._initial)), + rules_root=self._rules_root, + trace=self._trace, + walks=len(self._initial), + ) + except Exception: + return None diff --git a/src/sqlbuild/rule_engine/models.py b/src/sqlbuild/rule_engine/models.py index 040acdc066..0915f9ccfb 100644 --- a/src/sqlbuild/rule_engine/models.py +++ b/src/sqlbuild/rule_engine/models.py @@ -395,6 +395,23 @@ class CustomRuleEvaluation: reads: frozenset[tuple[object, ...]] | None +@dataclass(frozen=True) +class ModuleStateTrace: + """What exact module-state walks touched: object and type identities and their decisions.""" + + touched: dict[int, int] + descended_types: frozenset[int] + checked_instances: frozenset[int] + + +@dataclass(frozen=True) +class ModuleStateFingerprint: + """One exact custom-rule module-state fingerprint and the trace of the walk behind it.""" + + token: str + trace: ModuleStateTrace + + @dataclass(frozen=True) class CustomRuleRun: """Every invocation of one host run plus rules whose reads could not be attributed.""" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py index c08a3bdda2..a6ec47889b 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py @@ -201,3 +201,14 @@ class RulesCacheEditCase: edit: Callable[[Path], None] expected_exit_code: int = 1 expected_diagnostics_fragment: str = "" + + +@dataclass(frozen=True) +class SplitHostStateCase: + """Custom rules sharing helper tables, one with cross-model state, across host processes.""" + + description: str + model_count: int + selected_rules: tuple[str, ...] + files: tuple[tuple[str, str], ...] + expected_findings: tuple[tuple[str, str], ...] diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py index 80fa105a91..f961cc6a6c 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py @@ -5,6 +5,7 @@ import os import subprocess import sys +from collections.abc import Callable from pathlib import Path from typing import Any @@ -110,6 +111,7 @@ def run_compile_cli( *, environment: tuple[tuple[str, str], ...] = (), working_directory: Path | None = None, + preexec: Callable[[], None] | None = None, ) -> subprocess.CompletedProcess[str]: """Run the real `sqb compile --json` command from an optional invocation directory.""" @@ -127,9 +129,48 @@ def run_compile_cli( check=False, env={**os.environ, **dict(environment)}, cwd=working_directory, + preexec_fn=preexec, ) +def pin_to_one_cpu() -> None: + """Restrict the calling process, and so every custom-rule host it starts, to one CPU.""" + + os.sched_setaffinity(0, {min(os.sched_getaffinity(0))}) + + +SPLIT_ORDER_TABLES: str = ( + "import re\n\n" + "REGIONS = {f'region_{index:02d}': index for index in range(40)}\n" + "FLAGGED_SUFFIXES = ('3', '7')\n" + "ORDER_NAME = re.compile(r'^orders(_[0-9]{3})?$')\n" +) + + +def split_model_files(model_count: int) -> tuple[tuple[str, str], ...]: + """Return numbered order models beside the project's `models/orders.sql`.""" + + return tuple( + (f"models/orders_{index:03d}.sql", 'MODEL (description "Orders");\nSELECT 1 AS order_id\n') + for index in range(model_count) + ) + + +def split_expected_findings(model_count: int) -> tuple[tuple[str, str], ...]: + """Return what one host reports for the duplicate-parity and flagged-suffix order rules.""" + + names: tuple[str, ...] = tuple(f"orders_{index:03d}" for index in range(model_count)) + # `models/orders.sql` sorts first and is the first odd model; `orders_000` is the first even. + duplicates: tuple[tuple[str, str], ...] = tuple( + ("XSQBRSPLIT001", f"models/{name}.sql") for name in names[1:] + ) + suffixes: tuple[tuple[str, str], ...] = tuple( + ("XSQBRSPLIT002", f"models/{name}.sql") + for name in filter(lambda name: name.endswith(("3", "7")), names) + ) + return tuple(sorted(duplicates + suffixes)) + + def custom_rule_diagnostics(result: subprocess.CompletedProcess[str]) -> list[dict[str, Any]]: """Return custom-rule diagnostics from one JSON compile result.""" @@ -137,6 +178,14 @@ def custom_rule_diagnostics(result: subprocess.CompletedProcess[str]) -> list[di return list(filter(lambda item: str(item["code"]).startswith("XSQBR"), payload["diagnostics"])) +def custom_rule_findings(result: subprocess.CompletedProcess[str]) -> tuple[tuple[str, str], ...]: + """Return sorted custom-rule diagnostic codes and paths from one JSON compile result.""" + + return tuple( + sorted((str(item["code"]), str(item["path"])) for item in custom_rule_diagnostics(result)) + ) + + def custom_rule_codes(result: subprocess.CompletedProcess[str]) -> tuple[str, ...]: """Return custom-rule diagnostic codes from one JSON compile result.""" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_custom_rule_hermeticity.py b/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_custom_rule_hermeticity.py index 3b1486680c..0773c8b483 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_custom_rule_hermeticity.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_custom_rule_hermeticity.py @@ -13,16 +13,22 @@ NonHermeticRuleCase, ProjectTreeCacheCase, ScrubbedEnvironmentCase, + SplitHostStateCase, WorkingDirectoryCase, ) from tests.e2e.src.sqlbuild.cli.commands.main.rules.helpers import ( + SPLIT_ORDER_TABLES, custom_rule_codes, custom_rule_diagnostics, + custom_rule_findings, custom_rule_messages, custom_rule_paths, custom_rule_source, + pin_to_one_cpu, rule_cache_hits, run_compile_cli, + split_expected_findings, + split_model_files, string_set_order, write_custom_rule_project, ) @@ -52,6 +58,10 @@ ) +_SPLIT_MODEL_COUNT: int = 160 +_SPLIT_HEADER: str = "from sqlbuild.rules import Finding, Model, RuleContext, rule\n\nfrom rules import order_tables\n" + + @pytest.mark.parametrize( "test_case", [ @@ -491,3 +501,66 @@ def test_given_cached_rule_when_relative_helper_is_edited_then_rule_is_reevaluat assert custom_rule_codes(before) == test_case.expected_rule_codes_before_edit, before.stderr assert custom_rule_codes(after) == test_case.expected_rule_codes_after_edit, after.stderr + + +@pytest.mark.parametrize( + "test_case", + [ + SplitHostStateCase( + description="cross-model duplicate detector and stateless helper rule", + model_count=_SPLIT_MODEL_COUNT, + selected_rules=("XSQBRSPLIT001", "XSQBRSPLIT002"), + files=( + ("rules/__init__.py", ""), + ("rules/order_tables.py", SPLIT_ORDER_TABLES), + ( + "rules/duplicates.py", + custom_rule_source( + code="XSQBRSPLIT001", + header=_SPLIT_HEADER + "SEEN_PARITIES: set[str] = set()\n", + body=( + " even = model.name[-1:] in '02468'\n" + " parity = 'even' if even else 'odd'\n" + " duplicate = parity in SEEN_PARITIES\n" + " SEEN_PARITIES.add(parity)\n" + " matched = order_tables.ORDER_NAME.match(model.name)\n" + " return [ctx.finding(subject=model)] * int(duplicate and bool(matched))\n" + ), + ), + ), + ( + "rules/suffixes.py", + custom_rule_source( + code="XSQBRSPLIT002", + header=_SPLIT_HEADER, + body=( + " flagged = model.name.endswith(order_tables.FLAGGED_SUFFIXES)\n" + " known = len(order_tables.REGIONS) > 0\n" + " return [ctx.finding(subject=model)] * int(flagged and known)\n" + ), + ), + ), + *split_model_files(_SPLIT_MODEL_COUNT), + ), + expected_findings=split_expected_findings(_SPLIT_MODEL_COUNT), + ) + ], + ids=lambda case: case.description, +) +def test_given_stateful_and_stateless_rules_when_split_across_hosts_then_findings_match_one_host( + tmp_path: Path, test_case: SplitHostStateCase +) -> None: + single_dir: Path = tmp_path / "single" + split_dir: Path = tmp_path / "split" + for project_dir in (single_dir, split_dir): + project_dir.mkdir() + write_custom_rule_project( + project_dir=project_dir, selected_rules=test_case.selected_rules, files=test_case.files + ) + + single: subprocess.CompletedProcess[str] = run_compile_cli(single_dir, preexec=pin_to_one_cpu) + split: subprocess.CompletedProcess[str] = run_compile_cli(split_dir) + + assert (single.returncode, split.returncode) == (1, 1), single.stderr + split.stderr + assert custom_rule_findings(split) == custom_rule_findings(single) + assert custom_rule_findings(single) == test_case.expected_findings diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/host/_test_types.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/host/_test_types.py index 6ae0c71543..516e7d4943 100644 --- a/tests/unit/src/sqlbuild/rule_engine/_helpers/host/_test_types.py +++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/host/_test_types.py @@ -58,3 +58,33 @@ class HostCapacityTestCase: proc_cgroup: str files: dict[str, str] expected_cores: int + + +@dataclass(frozen=True) +class ModuleStateCase: + """Custom-rule modules whose state changes both detectors must classify identically.""" + + description: str + files: tuple[tuple[str, str], ...] + model_count: int + expected_stateful: frozenset[str] + + +@dataclass(frozen=True) +class RandomModuleStateCase: + """Seeded random rule mutations whose classification must match full re-fingerprinting.""" + + description: str + seed: int + model_count: int + expected_matches_reference: bool = True + + +@dataclass(frozen=True) +class StateDetectorCostCase: + """Stateless rules sharing helper tables, checked by both detectors after every call.""" + + description: str + rule_count: int + model_count: int + expected_min_speedup: float diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/host/helpers.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/host/helpers.py index cf8d7e815b..205e102db6 100644 --- a/tests/unit/src/sqlbuild/rule_engine/_helpers/host/helpers.py +++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/host/helpers.py @@ -1,10 +1,14 @@ """Builders for custom-rule host partitioning tests.""" import os +import random +import time from collections.abc import Callable from dataclasses import replace from operator import attrgetter from pathlib import Path +from types import SimpleNamespace +from typing import Protocol import pytest @@ -13,6 +17,9 @@ from sqlbuild.rule_engine._helpers.engine.catalogue import build_catalogue from sqlbuild.rule_engine._helpers.engine.custom_rules import evaluate_custom_rules_cached from sqlbuild.rule_engine._helpers.host import custom_host_pool +from sqlbuild.rule_engine._helpers.host.module_state import module_state_token, rule_namespaces +from sqlbuild.rule_engine.classes.module_state_monitor import ModuleStateMonitor +from sqlbuild.rule_engine.exceptions import OpaqueModuleStateError from sqlbuild.rule_engine.models import CustomRulesOutcome, Rule, RulesCacheConfig, RulesConfig from tests.unit.src.sqlbuild.rule_engine.main.evaluate.helpers import build_project @@ -136,3 +143,289 @@ def record(pid: int, signal_number: int) -> None: monkeypatch.setattr(os, "kill", record) monkeypatch.setattr(os, "killpg", record, raising=False) return signals + + +STATE_RULE_HEADER: str = "from sqlbuild.rules import Finding, Model, RuleContext, rule\n" +IMPORT_ORDER_HELPERS: str = "from rules import helpers\n" +ORDER_HELPERS: str = """import functools +import re + +STATUS_LABELS = {f"status_{index:03d}": f"Label {index}" for index in range(40)} +REGION_CODES = tuple(f"region_{index:03d}" for index in range(40)) +ALLOWED_PREFIXES = frozenset({"orders", "customers", "products"}) +PATTERNS = tuple(re.compile(rf"^orders_[0-9]+{index}$") for index in range(5)) +NESTED = {f"group_{index}": tuple((f"key_{key}", key) for key in range(4)) for index in range(5)} +SHARED: dict[str, int] = {} + + +class Registry: + entries = [] + + +def register(name: str) -> None: + Registry.entries.append(name) + + +@functools.lru_cache(maxsize=None) +def cached_length(name: str) -> int: + return len(name) + + +def is_known(name: str) -> bool: + return name in STATUS_LABELS or any(pattern.match(name) for pattern in PATTERNS) +""" +RAISING_PROXY_LIBRARY: str = """class BoomError(Exception): + pass + + +class Proxy: + def __getattr__(self, name): + raise BoomError(name) + + +class Helper: + proxy = Proxy() + + +def helper_fn(): + return 1 +""" +SIDE_EFFECT_PROPERTY_LIBRARY: str = """CALLS = [] + + +class Lazy: + @property + def cache_info(self): + CALLS.append(1) + raise AttributeError("cache_info") + + +class Holder: + lazy = Lazy() + + +def lazy_fn(): + return len(CALLS) +""" +FOREIGN_MODULE_CLASS_LIBRARY: str = """class BoomError(Exception): + pass + + +class Proxy: + def __getattr__(self, name): + raise BoomError(name) + + +class Hidden: + proxy = Proxy() + + +Hidden.__module__ = "rules.cross_b" +""" +STATELESS_RULE_CODE: str = "XSQBRST900" +STATEFUL_RULE_CODE: str = "XSQBRST001" +RANDOM_OPERATIONS: tuple[str, ...] = ( + "_ = helpers.is_known(model.name)", + "LOCAL[model.name] = 1", + "ITEMS.append(model.index)", + "ITEMS.append(1)\nITEMS.pop()", + "if ITEMS:\n ITEMS.pop()\nelse:\n ITEMS.append(1)", + "MARKS.add(model.name)", + "Holder.total += 1", + "BOX.value = model.index", + "globals()['COUNTER'] += 1", + "globals()['TABLE'] = dict(TABLE)", + "NESTED['inner'].append(model.index)", + "helpers.SHARED[model.name] = 1", + "helpers.register(model.name)", + "_ = helpers.cached_length(model.name)", +) +RANDOM_HEADERS: tuple[str, ...] = ( + "from rules.helpers import Registry\n" + IMPORT_ORDER_HELPERS, + IMPORT_ORDER_HELPERS + "from rules.helpers import Registry\n", +) +RANDOM_STATE: str = ( + "LOCAL: dict[str, int] = {}\nITEMS: list[int] = []\nMARKS: set[str] = set()\n" + "COUNTER = 0\nTABLE = {'orders': 1}\nNESTED: dict[str, list[int]] = {'inner': []}\n\n\n" + "class Holder:\n total = 0\n\n\nclass _Box:\n value = 0\n\n\nBOX = _Box()\n" +) + + +def state_rule(*, code: str, body: str, header: str = "", function: str = "check") -> str: + """Return one model-subject custom Rule module whose check runs `body` and finds nothing.""" + + indented: str = "".join(f" {line}\n" for line in body.splitlines()) + return ( + f"{STATE_RULE_HEADER}{header}\n\n" + f'@rule(code="{code}", message="orders state", remediation="Keep rules pure.")\n' + f"def {function}(*, model: Model, ctx: RuleContext) -> list[Finding]:\n" + f"{indented} return []\n" + ) + + +def order_helpers_file() -> tuple[str, str]: + """Return the shared helper module with constant tables, a registry, and an lru cache.""" + + return ("rules/helpers.py", ORDER_HELPERS) + + +def stateless_rule_file(code: str = STATELESS_RULE_CODE) -> tuple[str, str]: + """Return a rule that only reads the shared helper tables.""" + + return ( + f"rules/{code.lower()}.py", + state_rule( + code=code, + header=IMPORT_ORDER_HELPERS + "LIMITS = {'orders': (1, 2, 3)}\n", + body="_ = helpers.is_known(model.name) and model.name in helpers.REGION_CODES", + ), + ) + + +def state_case_files(*, header: str, body: str) -> tuple[tuple[str, str], ...]: + """Return the helper, one rule running `body` under `header`, and one stateless rule.""" + + return ( + order_helpers_file(), + ("rules/orders_state.py", state_rule(code=STATEFUL_RULE_CODE, header=header, body=body)), + stateless_rule_file(), + ) + + +def random_state_files(*, seed: int, model_count: int) -> tuple[tuple[str, str], ...]: + """Return two to five rules each running one seeded mutation for a seeded set of models.""" + + generator: random.Random = random.Random(seed) + files: list[tuple[str, str]] = [order_helpers_file()] + for index in range(generator.randint(2, 5)): + operation: str = generator.choice(RANDOM_OPERATIONS) + triggers: list[int] = sorted(generator.sample(range(model_count), generator.randint(0, 3))) + indented: str = "".join(f" {line}\n" for line in operation.splitlines()) + files.append( + ( + f"rules/random_{index}.py", + state_rule( + code=f"XSQBRSR{index:03d}", + header=generator.choice(RANDOM_HEADERS) + RANDOM_STATE, + body=f"if model.index in {triggers!r}:\n{indented}", + ), + ) + ) + return tuple(files) + + +class StateDetector(Protocol): + """A module-state detector reporting newly stateful codes after each call.""" + + def changed_codes(self) -> tuple[str, ...]: ... + + +class ReferenceStateDetector: + """The detector before CHI-491: re-fingerprint every remaining code after each call.""" + + def __init__( + self, *, namespaces: dict[str, tuple[dict[str, object], ...]], rules_root: Path + ) -> None: + self._namespaces: dict[str, tuple[dict[str, object], ...]] = namespaces + self._rules_root: Path = rules_root + self._initial: dict[str, str | None] = {code: self._token(code) for code in namespaces} + self._pending: tuple[str, ...] = tuple(namespaces) + + def changed_codes(self) -> tuple[str, ...]: + tokens: dict[str, str | None] = {code: self._token(code) for code in self._pending} + changed: tuple[str, ...] = tuple( + filter( + lambda code: tokens[code] is None or tokens[code] != self._initial[code], + self._pending, + ) + ) + self._pending = tuple(filter(lambda code: code not in changed, self._pending)) + return changed + + def _token(self, code: str) -> str | None: + try: + return module_state_token( + namespaces=self._namespaces[code], rules_root=self._rules_root + ) + except OpaqueModuleStateError: + return None + + +def load_state_rules(*, root: Path, files: tuple[tuple[str, str], ...]) -> tuple[Rule, ...]: + """Write custom-rule files beneath `root` and load every custom rule they define.""" + + # A packaged `rules` unloads with the project, so no later test imports a stale helper module. + for relative_path, contents in (("rules/__init__.py", ""), *files): + path: Path = root / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(contents, encoding="utf-8") + return tuple( + filter(attrgetter("custom"), build_catalogue(config=RulesConfig(), project_dir=root)) + ) + + +def _rule_namespaces(rules: tuple[Rule, ...]) -> dict[str, tuple[dict[str, object], ...]]: + return {rule.code: rule_namespaces((rule.check,)) for rule in rules} + + +def reference_detector(*, rules: tuple[Rule, ...], root: Path) -> StateDetector: + """Return the full re-fingerprinting reference detector for `rules`.""" + + return ReferenceStateDetector(namespaces=_rule_namespaces(rules), rules_root=root / "rules") + + +def monitor_detector(*, rules: tuple[Rule, ...], root: Path) -> StateDetector: + """Return the production module-state monitor for `rules`.""" + + return ModuleStateMonitor(namespaces=_rule_namespaces(rules), rules_root=root / "rules") + + +def state_histories( + *, rules: tuple[Rule, ...], model_count: int, detectors: tuple[StateDetector, ...] +) -> tuple[list[tuple[str, ...]], ...]: + """Run every rule over numbered order models and record each detector after each call.""" + + histories: tuple[list[tuple[str, ...]], ...] = tuple([] for _ in detectors) + for index in range(model_count): + subject: SimpleNamespace = SimpleNamespace(name=f"orders_{index:03d}", index=index) + for rule in rules: + _ = rule.check( + **{str(rule.subject_parameter): subject, str(rule.context_parameter): None} + ) + for history, detector in zip(histories, detectors, strict=True): + history.append(tuple(sorted(detector.changed_codes()))) + return histories + + +def compare_state_detectors( + *, root: Path, files: tuple[tuple[str, str], ...], model_count: int +) -> tuple[list[tuple[str, ...]], list[tuple[str, ...]]]: + """Return the reference and monitor histories observed side by side over the same calls.""" + + rules: tuple[Rule, ...] = load_state_rules(root=root, files=files) + reference, monitor = state_histories( + rules=rules, + model_count=model_count, + detectors=( + reference_detector(rules=rules, root=root), + monitor_detector(rules=rules, root=root), + ), + ) + return reference, monitor + + +def timed_state_history( + *, + root: Path, + files: tuple[tuple[str, str], ...], + model_count: int, + detector: Callable[..., StateDetector], +) -> tuple[float, list[tuple[str, ...]]]: + """Return the CPU seconds one detector spends over every call, and its history.""" + + rules: tuple[Rule, ...] = load_state_rules(root=root, files=files) + started: float = time.process_time() + (history,) = state_histories( + rules=rules, model_count=model_count, detectors=(detector(rules=rules, root=root),) + ) + return time.process_time() - started, history diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/host/test_module_state.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/host/test_module_state.py new file mode 100644 index 0000000000..6036adc7b3 --- /dev/null +++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/host/test_module_state.py @@ -0,0 +1,427 @@ +"""Custom-rule module-state detection classifies every rule exactly as full re-fingerprinting.""" + +from itertools import chain +from pathlib import Path + +import pytest + +from tests.unit.src.sqlbuild.rule_engine._helpers.host._test_types import ( + ModuleStateCase, + RandomModuleStateCase, + StateDetectorCostCase, +) +from tests.unit.src.sqlbuild.rule_engine._helpers.host.helpers import ( + FOREIGN_MODULE_CLASS_LIBRARY, + IMPORT_ORDER_HELPERS, + RAISING_PROXY_LIBRARY, + SIDE_EFFECT_PROPERTY_LIBRARY, + STATEFUL_RULE_CODE, + STATELESS_RULE_CODE, + compare_state_detectors, + monitor_detector, + order_helpers_file, + random_state_files, + reference_detector, + state_case_files, + state_rule, + stateless_rule_file, + timed_state_history, +) + +_STATEFUL: frozenset[str] = frozenset({STATEFUL_RULE_CODE}) +_STATELESS: frozenset[str] = frozenset() +_ISOLATED_RULE: tuple[str, str] = ( + "rules/isolated.py", + state_rule(code="XSQBRST902", body="_ = model.name"), +) + + +@pytest.mark.parametrize( + "test_case", + [ + ModuleStateCase( + description="stateless constant tables", + files=state_case_files( + header=IMPORT_ORDER_HELPERS + "SIZES = {'small': 1}\nNAMES = ('orders', 'items')\n", + body="_ = helpers.is_known(model.name) or SIZES.get(model.name) or NAMES", + ), + model_count=4, + expected_stateful=_STATELESS, + ), + ModuleStateCase( + description="dict mutation", + files=state_case_files( + header="SEEN: dict[str, bool] = {}\n", body="SEEN[model.name] = True" + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="list append", + files=state_case_files(header="ITEMS: list[int] = []\n", body="ITEMS.append(1)"), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="set add", + files=state_case_files( + header="MARKS: set[str] = set()\n", body="MARKS.add(model.name)" + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="list element replaced in place", + files=state_case_files(header="ITEMS = ['orders', 1]\n", body="ITEMS[0] = model.name"), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="dict value replaced without resizing", + files=state_case_files( + header="LABELS = {'orders': 'Orders', 'customers': 2}\n", + body="LABELS['orders'] = model.name", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="set member swapped without resizing", + files=state_case_files( + header="MARKS = {'orders', 'customers'}\n", + body="MARKS.discard('orders')\nMARKS.add(model.name)", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="attribute on module-level object", + files=state_case_files( + header=( + "class Box:\n def __init__(self) -> None:\n self.value = ''\n\n\n" + "BOX = Box()\n" + ), + body="BOX.value = model.name", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="class attribute", + files=state_case_files( + header="class Totals:\n count = 0\n", body="Totals.count += 1" + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="global counter rebinding", + files=state_case_files(header="COUNTER = 0\n", body="globals()['COUNTER'] += 1"), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="rebinding a table to an equal copy", + files=state_case_files( + header="TABLE = {'orders': 1}\n", body="globals()['TABLE'] = dict(TABLE)" + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="mutate and restore within one call", + files=state_case_files( + header="ITEMS: list[int] = []\n", body="ITEMS.append(1)\nITEMS.pop()" + ), + model_count=4, + expected_stateful=_STATELESS, + ), + ModuleStateCase( + description="mutate on one call and restore on the next", + files=state_case_files( + header="TOGGLE: list[int] = []\n", + body="if TOGGLE:\n TOGGLE.pop()\nelse:\n TOGGLE.append(1)", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="lazily initialised cache", + files=state_case_files( + header="CACHE: dict[str, int] | None = None\n", + body="if CACHE is None:\n globals()['CACHE'] = {'orders': 1}", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="nested containers", + files=state_case_files( + header="NESTED = {'groups': [{'names': ['orders']}]}\n", + body="NESTED['groups'][0]['names'].append(model.name)", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="cycle without mutation", + files=state_case_files( + header="LOOP: list[object] = []\nLOOP.append(LOOP)\n", body="_ = LOOP[0] is LOOP" + ), + model_count=4, + expected_stateful=_STATELESS, + ), + ModuleStateCase( + description="mutation inside a cycle", + files=state_case_files( + header="LOOP: list[object] = []\nLOOP.append(LOOP)\n", + body="LOOP[0].append(model.name)", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="closure cell", + files=state_case_files( + header=( + "def _counter():\n total = 0\n\n def bump() -> int:\n" + " nonlocal total\n total += 1\n return total\n\n" + " return bump\n\n\nBUMP = _counter()\n" + ), + body="_ = BUMP()", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="opaque lock state", + files=state_case_files( + header="import threading\n\nLOCK = threading.Lock()\n", body="_ = model.name" + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="state too large to observe", + files=state_case_files(header="LARGE = list(range(60_000))\n", body="_ = model.name"), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="dataclass instance from the rules module", + files=state_case_files( + header=( + "from dataclasses import dataclass, field\n\n\n@dataclass\nclass Seen:\n" + " names: list[str] = field(default_factory=list)\n\n\nSEEN = Seen()\n" + ), + body="SEEN.names.append(model.name)", + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="mutation after several stateless calls", + files=state_case_files( + header="LATE: list[str] = []\n", + body="if model.index == 3:\n LATE.append(model.name)", + ), + model_count=5, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="bytearray mutated in place", + files=state_case_files( + header="BUFFER = bytearray(b'orders')\n", body="BUFFER[0] = model.index" + ), + model_count=3, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="float signed zero", + files=state_case_files(header="RATIO = 0.0\n", body="globals()['RATIO'] = -0.0"), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="mutation in a shared helper reaches every importing rule", + files=( + *state_case_files( + header=IMPORT_ORDER_HELPERS, body="helpers.SHARED[model.name] = 1" + ), + _ISOLATED_RULE, + ), + model_count=3, + expected_stateful=frozenset({STATEFUL_RULE_CODE, STATELESS_RULE_CODE}), + ), + ModuleStateCase( + description="helper lru cache reaches every importing rule", + files=( + *state_case_files( + header=IMPORT_ORDER_HELPERS, body="_ = helpers.cached_length(model.name)" + ), + _ISOLATED_RULE, + ), + model_count=3, + expected_stateful=frozenset({STATEFUL_RULE_CODE, STATELESS_RULE_CODE}), + ), + ModuleStateCase( + description="helper class reached before its module is not observed", + files=( + order_helpers_file(), + ( + "rules/orders_state.py", + state_rule( + code=STATEFUL_RULE_CODE, + header=IMPORT_ORDER_HELPERS, + body="helpers.register(model.name)", + ), + ), + ( + "rules/registry_reader.py", + state_rule( + code="XSQBRST903", + header="from rules.helpers import Registry\n" + IMPORT_ORDER_HELPERS, + body="_ = Registry.entries or helpers.REGION_CODES", + ), + ), + ), + model_count=4, + expected_stateful=_STATEFUL, + ), + ModuleStateCase( + description="helper class with a raising proxy reached before its module", + files=( + ("rules/lib.py", RAISING_PROXY_LIBRARY), + ( + "rules/a.py", + state_rule( + code="XSQBRST905", + header="from rules.lib import Helper, helper_fn\n", + body="_ = helper_fn() and Helper", + ), + ), + ), + model_count=3, + expected_stateful=_STATELESS, + ), + ModuleStateCase( + description="side-effecting property in a class no fingerprint descends into", + files=( + ("rules/lazy.py", SIDE_EFFECT_PROPERTY_LIBRARY), + ( + "rules/a.py", + state_rule( + code="XSQBRST906", + header="from rules.lazy import Holder, lazy_fn\n", + body="_ = lazy_fn() or Holder", + ), + ), + ), + model_count=3, + expected_stateful=_STATELESS, + ), + ModuleStateCase( + description="class from another rule's module that its namespace does not hold", + files=( + ("rules/parts.py", FOREIGN_MODULE_CLASS_LIBRARY), + ( + "rules/cross_a.py", + state_rule( + code="XSQBRST907", + header="from rules.parts import Hidden\n", + body="_ = Hidden.__name__", + ), + ), + ("rules/cross_b.py", state_rule(code="XSQBRST908", body="_ = model.name")), + ), + model_count=3, + expected_stateful=_STATELESS, + ), + ModuleStateCase( + description="two rules in one module share their classification", + files=( + order_helpers_file(), + ( + "rules/orders_state.py", + state_rule( + code=STATEFUL_RULE_CODE, + header="SEEN: list[str] = []\n", + body="SEEN.append(model.name)", + ) + + state_rule(code="XSQBRST904", body="_ = model.name", function="second"), + ), + stateless_rule_file(), + ), + model_count=3, + expected_stateful=frozenset({STATEFUL_RULE_CODE, "XSQBRST904"}), + ), + ], + ids=lambda case: case.description, +) +def test_given_rule_module_state_when_monitoring_then_matches_full_refingerprinting( + tmp_path: Path, test_case: ModuleStateCase +) -> None: + expected, observed = compare_state_detectors( + root=tmp_path, files=test_case.files, model_count=test_case.model_count + ) + + assert observed == expected + assert frozenset(chain.from_iterable(observed)) == test_case.expected_stateful + + +@pytest.mark.parametrize( + "test_case", + [ + RandomModuleStateCase(description=f"seed {seed}", seed=seed, model_count=6) + for seed in range(40) + ], + ids=lambda case: case.description, +) +def test_given_random_rule_mutations_when_monitoring_then_matches_full_refingerprinting( + tmp_path: Path, test_case: RandomModuleStateCase +) -> None: + expected, observed = compare_state_detectors( + root=tmp_path, + files=random_state_files(seed=test_case.seed, model_count=test_case.model_count), + model_count=test_case.model_count, + ) + + assert (observed == expected) is test_case.expected_matches_reference + + +@pytest.mark.parametrize( + "test_case", + [ + StateDetectorCostCase( + description="twelve stateless rules sharing helper tables", + rule_count=12, + model_count=25, + expected_min_speedup=4.0, + ) + ], + ids=lambda case: case.description, +) +def test_given_rules_sharing_helper_tables_when_monitoring_then_cheaper_than_refingerprinting( + tmp_path: Path, test_case: StateDetectorCostCase +) -> None: + files: tuple[tuple[str, str], ...] = ( + order_helpers_file(), + *(stateless_rule_file(code=f"XSQBRSP{index:03d}") for index in range(test_case.rule_count)), + ) + + reference_seconds, reference_history = timed_state_history( + root=tmp_path / "reference", + files=files, + model_count=test_case.model_count, + detector=reference_detector, + ) + monitor_seconds, monitor_history = timed_state_history( + root=tmp_path / "monitor", + files=files, + model_count=test_case.model_count, + detector=monitor_detector, + ) + + assert monitor_history == reference_history + assert not any(monitor_history) + assert monitor_seconds * test_case.expected_min_speedup < reference_seconds