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