msyavuz commented on code in PR #44892:
URL: https://github.com/apache/superset/pull/44892#discussion_r4195219815
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +913,349 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...] | None:
+ """Materialize the provider's payload, or reject it.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return None
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ return tuple(provided)
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s could not be read; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
+
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
+ try:
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Both resolve from core's own identities: the stock listener
+ # actions decide what to delete from a built-in entity type, and
+ # blockers are applied by the stock validator. Declared by a host,
+ # they are accepted and then never carried out -- silence rather
+ # than a failure, which is the one outcome worth refusing. A host
+ # blocks by raising PurgeBlockedError from its own validate.
+ raise RuntimeError(
Review Comment:
**major**: Rejecting BLOCK and LISTENER_EFFECT stops a host from declaring
them, but the built-in stock cleanup still runs on host roots. Tag, permission
and other core listener cleanup keyed off entity_type won't run, so host rows
can be purged with core-side references left dangling and no error.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +913,349 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...] | None:
+ """Materialize the provider's payload, or reject it.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return None
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ return tuple(provided)
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s could not be read; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
+
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
+ try:
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Both resolve from core's own identities: the stock listener
+ # actions decide what to delete from a built-in entity type, and
+ # blockers are applied by the stock validator. Declared by a host,
+ # they are accepted and then never carried out -- silence rather
+ # than a failure, which is the one outcome worth refusing. A host
+ # blocks by raising PurgeBlockedError from its own validate.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies cannot be carried
"
+ "out for a host root"
+ )
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...] | None:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+
+ Returns ``None`` for the one failure that may be transient -- the provider
+ itself could not be called. The caller then serves the built-in roots
+ *without* publishing, so the next resolution tries again instead of
+ freezing a bad moment in for the life of the process. A payload that is
+ merely malformed is a host bug that will not fix itself, so it is
+ published as an empty result rather than retried on every read.
+ """
+ if provider is None:
+ return ()
+ try:
+ provided: Any = provider()
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+ candidates: tuple[Any, ...] | None = _host_policy_payload(provided)
+ if candidates is None:
+ return ()
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
+
+
+#: Sentinel distinguishing "no index resolved yet" from an index resolved for
+#: no provider, which is the ordinary case.
+_UNRESOLVED: Any = object()
+
+
+@dataclass
+class _ResolvedRegistry:
+ """The index built for one provider, with the roots validated so far.
+
+ Compared by provider *identity* rather than memoized with ``lru_cache``: a
+ cache keyed on the provider would hash it, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That would
+ raise here -- before the host boundary could isolate the host's mistake --
+ and stop the built-in roots purging too.
+ """
+
+ provider: Any = _UNRESOLVED
+ registry: Mapping[type[Any], PurgeEntityPolicy] = field(
+ default_factory=lambda: MappingProxyType({})
+ )
+ validated: set[type[Any]] = field(default_factory=set)
+
+
+_RESOLVED: _ResolvedRegistry = _ResolvedRegistry()
+
+#: Per-thread marker for "this thread is already resolving". A host provider
+#: may build its policy from a built-in one -- ``replace(get_purge_policy(
+#: Slice), model=HostRoot, ...)`` is the obvious way to borrow the stock
+#: callbacks -- which re-enters resolution while the first call is still in
+#: the provider. The nested call is handed the state already published
+#: instead, which always carries the built-in roots.
+_RESOLVING: threading.local = threading.local()
+
+
+def _resolved_registry() -> _ResolvedRegistry:
+ """Return the state resolved for the installed provider, building it once.
+
+ Readers take a snapshot and a rebuild publishes a new one, so a caller
+ always sees an index and its validation record together. Publishing is a
+ single rebind, which is what makes it safe to do this without a lock: two
+ concurrent first uses may each invoke the provider and each publish, but
+ the two snapshots are equivalent and neither is ever seen half-built.
+ Paying for a duplicate provider call is the deliberate trade -- a lock
+ here deadlocks the moment a provider resolves a built-in policy of its
+ own, and a hung purge is worse than a repeated call.
+ """
+ provider: Callable[[], Any] | None = (
+ current_app.config.get(HOST_POLICIES_CONFIG_KEY) if has_app_context()
else None
+ )
+ resolved: _ResolvedRegistry = _RESOLVED
+ if resolved.provider is provider:
+ return resolved
+ if getattr(_RESOLVING, "active", False):
+ # A nested resolution: the provider is building its policy from a
+ # built-in one. It needs the roots this package declares -- not
+ # whatever happens to be published, which on a first resolution is
+ # still the empty initial snapshot.
+ return _builtin_registry()
Review Comment:
**major**: A nested resolution returns `_builtin_registry()`, which is
`lru_cache`d and carries `provider=_UNRESOLVED`. A provider that calls
`get_purge_policy` on a built-in root therefore validates against a separate
`validated` set. Together with the unpublished transient-failure path, this
means built-in roots are validated twice and the two snapshots can disagree.
##########
superset/tasks/deletion_retention.py:
##########
@@ -192,44 +244,61 @@ def _purge_impl(window_days: int, dry_run: bool) ->
dict[str, Any]:
stats_logger_manager.instance.gauge(
f"{_METRIC_PREFIX}.blocked_by_reference", blocked
)
+ if scan_failures:
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.scan_failures", scan_failures
+ )
stats = {
"purged": purged,
"cascade_failures": failures,
"blocked_by_reference": blocked,
"unsupported_models": unsupported_models,
+ "scan_failures": scan_failures,
}
logger.info("deletion_retention: %s", stats)
return stats
def _purge_model(
model: type[SoftDeleteMixin], cutoff: datetime, dry_run: bool
-) -> tuple[int, int, int, int]:
+) -> tuple[int, int, int, int, int]:
"""Process one model's eligible rows. Returns ``(purged, would_purge,
- failures, blocked)``. A single entity's blocked/failed cascade never aborts
- the batch."""
+ failures, blocked, scan_failures)``. A single entity's blocked/failed
+ cascade never aborts the batch.
+
+ A failure in the eligible-id scan itself -- a column it cannot read, a
+ transient database error between pages -- ends this model's pass and is
+ reported, but the counts earned before it are kept: by then those rows are
+ committed deletions, and a summary that omitted them would understate what
+ the run actually removed.
+ """
entity_type = _model_table_name(model)
- purged = would = failures = blocked = 0
- for id_batch in _iter_eligible_ids(model, cutoff, _BATCH):
- if dry_run:
- would += len(id_batch)
- continue
- for entity_id in id_batch:
- try:
- result = _purge_one(model, entity_id, cutoff)
- if result is not None and result.purged:
- purged += 1
- elif result is not None and result.blocked_reason is not None:
- blocked += 1
- except Exception: # pylint: disable=broad-except
- db.session.rollback() # pylint:
disable=consider-using-transaction
- failures += 1
- logger.exception(
- "deletion_retention: cascade failed for %s id=%s",
- entity_type,
- entity_id,
- )
- return purged, would, failures, blocked
+ purged = would = failures = blocked = scan_failures = 0
+ try:
+ for id_batch in _iter_eligible_ids(model, cutoff, _BATCH):
+ if dry_run:
+ would += len(id_batch)
+ continue
+ for entity_id in id_batch:
+ try:
+ result = _purge_one(model, entity_id, cutoff)
+ if result is not None and result.purged:
+ purged += 1
+ elif result is not None and result.blocked_reason is not
None:
+ blocked += 1
+ except Exception: # pylint: disable=broad-except
+ db.session.rollback() # pylint:
disable=consider-using-transaction
+ failures += 1
+ logger.exception(
+ "deletion_retention: cascade failed for %s id=%s",
+ entity_type,
+ entity_id,
+ )
+ except Exception: # pylint: disable=broad-except
Review Comment:
**minor**: The scan-failure handler rolls back the session after earlier
pages have committed, then moves on. A failure repeated every night for one
root only shows up as a gauge, and the run still returns normally. Nothing in
the task's return status or exit code flags it.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +913,349 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...] | None:
+ """Materialize the provider's payload, or reject it.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return None
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ return tuple(provided)
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s could not be read; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
+
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
+ try:
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Both resolve from core's own identities: the stock listener
+ # actions decide what to delete from a built-in entity type, and
+ # blockers are applied by the stock validator. Declared by a host,
+ # they are accepted and then never carried out -- silence rather
+ # than a failure, which is the one outcome worth refusing. A host
+ # blocks by raising PurgeBlockedError from its own validate.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies cannot be carried
"
+ "out for a host root"
+ )
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...] | None:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+
+ Returns ``None`` for the one failure that may be transient -- the provider
+ itself could not be called. The caller then serves the built-in roots
+ *without* publishing, so the next resolution tries again instead of
+ freezing a bad moment in for the life of the process. A payload that is
+ merely malformed is a host bug that will not fix itself, so it is
+ published as an empty result rather than retried on every read.
+ """
+ if provider is None:
+ return ()
+ try:
+ provided: Any = provider()
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+ candidates: tuple[Any, ...] | None = _host_policy_payload(provided)
+ if candidates is None:
+ return ()
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
Review Comment:
**minor**: Host-supplied `delete_*` callbacks are executed inside the locked
purge frame and are not checked at admission. A host that cloned the chart
policy via `replace(...)` without overriding them would run chart-specific
cleanup against its own table. Only the reserved `entity_type` is guarded.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +913,349 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...] | None:
+ """Materialize the provider's payload, or reject it.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return None
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ return tuple(provided)
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s could not be read; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
+
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
+ try:
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Both resolve from core's own identities: the stock listener
+ # actions decide what to delete from a built-in entity type, and
+ # blockers are applied by the stock validator. Declared by a host,
+ # they are accepted and then never carried out -- silence rather
+ # than a failure, which is the one outcome worth refusing. A host
+ # blocks by raising PurgeBlockedError from its own validate.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies cannot be carried
"
+ "out for a host root"
+ )
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...] | None:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+
+ Returns ``None`` for the one failure that may be transient -- the provider
+ itself could not be called. The caller then serves the built-in roots
+ *without* publishing, so the next resolution tries again instead of
+ freezing a bad moment in for the life of the process. A payload that is
+ merely malformed is a host bug that will not fix itself, so it is
+ published as an empty result rather than retried on every read.
+ """
+ if provider is None:
+ return ()
+ try:
+ provided: Any = provider()
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+ candidates: tuple[Any, ...] | None = _host_policy_payload(provided)
+ if candidates is None:
+ return ()
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
+
+
+#: Sentinel distinguishing "no index resolved yet" from an index resolved for
+#: no provider, which is the ordinary case.
+_UNRESOLVED: Any = object()
+
+
+@dataclass
+class _ResolvedRegistry:
+ """The index built for one provider, with the roots validated so far.
+
+ Compared by provider *identity* rather than memoized with ``lru_cache``: a
+ cache keyed on the provider would hash it, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That would
+ raise here -- before the host boundary could isolate the host's mistake --
+ and stop the built-in roots purging too.
+ """
+
+ provider: Any = _UNRESOLVED
+ registry: Mapping[type[Any], PurgeEntityPolicy] = field(
+ default_factory=lambda: MappingProxyType({})
+ )
+ validated: set[type[Any]] = field(default_factory=set)
+
+
+_RESOLVED: _ResolvedRegistry = _ResolvedRegistry()
+
+#: Per-thread marker for "this thread is already resolving". A host provider
+#: may build its policy from a built-in one -- ``replace(get_purge_policy(
+#: Slice), model=HostRoot, ...)`` is the obvious way to borrow the stock
+#: callbacks -- which re-enters resolution while the first call is still in
+#: the provider. The nested call is handed the state already published
+#: instead, which always carries the built-in roots.
+_RESOLVING: threading.local = threading.local()
+
+
+def _resolved_registry() -> _ResolvedRegistry:
+ """Return the state resolved for the installed provider, building it once.
+
+ Readers take a snapshot and a rebuild publishes a new one, so a caller
+ always sees an index and its validation record together. Publishing is a
+ single rebind, which is what makes it safe to do this without a lock: two
+ concurrent first uses may each invoke the provider and each publish, but
+ the two snapshots are equivalent and neither is ever seen half-built.
+ Paying for a duplicate provider call is the deliberate trade -- a lock
+ here deadlocks the moment a provider resolves a built-in policy of its
+ own, and a hung purge is worse than a repeated call.
+ """
+ provider: Callable[[], Any] | None = (
+ current_app.config.get(HOST_POLICIES_CONFIG_KEY) if has_app_context()
else None
+ )
+ resolved: _ResolvedRegistry = _RESOLVED
+ if resolved.provider is provider:
Review Comment:
**major**: Cache identity is the provider object, so a provider that returns
a fresh callable (a lambda built per call, say) or a config toggled between app
contexts rebuilds on every call. In a multi-app process it also thrashes the
single global snapshot, and the provider then runs once per root.
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +913,349 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
+
+
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...] | None:
+ """Materialize the provider's payload, or reject it.
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return None
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ return tuple(provided)
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s could not be read; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
+
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
+ try:
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ unusable: set[str] = {
+ dependency.classification.value
+ for dependency in candidate.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.BLOCK,
+ DependencyClassification.LISTENER_EFFECT,
+ }
+ }
+ if unusable:
+ # Both resolve from core's own identities: the stock listener
+ # actions decide what to delete from a built-in entity type, and
+ # blockers are applied by the stock validator. Declared by a host,
+ # they are accepted and then never carried out -- silence rather
+ # than a failure, which is the one outcome worth refusing. A host
+ # blocks by raising PurgeBlockedError from its own validate.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies cannot be carried
"
+ "out for a host root"
+ )
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...] | None:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+
+ Returns ``None`` for the one failure that may be transient -- the provider
+ itself could not be called. The caller then serves the built-in roots
+ *without* publishing, so the next resolution tries again instead of
+ freezing a bad moment in for the life of the process. A payload that is
+ merely malformed is a host bug that will not fix itself, so it is
+ published as an empty result rather than retried on every read.
+ """
+ if provider is None:
+ return ()
+ try:
+ provided: Any = provider()
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+ candidates: tuple[Any, ...] | None = _host_policy_payload(provided)
+ if candidates is None:
+ return ()
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
+
+
+#: Sentinel distinguishing "no index resolved yet" from an index resolved for
+#: no provider, which is the ordinary case.
+_UNRESOLVED: Any = object()
+
+
+@dataclass
+class _ResolvedRegistry:
+ """The index built for one provider, with the roots validated so far.
+
+ Compared by provider *identity* rather than memoized with ``lru_cache``: a
+ cache keyed on the provider would hash it, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That would
+ raise here -- before the host boundary could isolate the host's mistake --
+ and stop the built-in roots purging too.
+ """
+
+ provider: Any = _UNRESOLVED
+ registry: Mapping[type[Any], PurgeEntityPolicy] = field(
+ default_factory=lambda: MappingProxyType({})
+ )
+ validated: set[type[Any]] = field(default_factory=set)
+
+
+_RESOLVED: _ResolvedRegistry = _ResolvedRegistry()
+
+#: Per-thread marker for "this thread is already resolving". A host provider
+#: may build its policy from a built-in one -- ``replace(get_purge_policy(
+#: Slice), model=HostRoot, ...)`` is the obvious way to borrow the stock
+#: callbacks -- which re-enters resolution while the first call is still in
+#: the provider. The nested call is handed the state already published
+#: instead, which always carries the built-in roots.
+_RESOLVING: threading.local = threading.local()
+
+
+def _resolved_registry() -> _ResolvedRegistry:
+ """Return the state resolved for the installed provider, building it once.
+
+ Readers take a snapshot and a rebuild publishes a new one, so a caller
+ always sees an index and its validation record together. Publishing is a
+ single rebind, which is what makes it safe to do this without a lock: two
+ concurrent first uses may each invoke the provider and each publish, but
+ the two snapshots are equivalent and neither is ever seen half-built.
+ Paying for a duplicate provider call is the deliberate trade -- a lock
+ here deadlocks the moment a provider resolves a built-in policy of its
+ own, and a hung purge is worse than a repeated call.
+ """
+ provider: Callable[[], Any] | None = (
+ current_app.config.get(HOST_POLICIES_CONFIG_KEY) if has_app_context()
else None
+ )
+ resolved: _ResolvedRegistry = _RESOLVED
+ if resolved.provider is provider:
+ return resolved
+ if getattr(_RESOLVING, "active", False):
+ # A nested resolution: the provider is building its policy from a
+ # built-in one. It needs the roots this package declares -- not
+ # whatever happens to be published, which on a first resolution is
+ # still the empty initial snapshot.
+ return _builtin_registry()
+ _RESOLVING.active = True
+ try:
+ host_policies: tuple[PurgeEntityPolicy, ...] | None =
_host_purge_policies(
+ provider
+ )
+ finally:
+ _RESOLVING.active = False
+ if host_policies is None:
+ # The provider could not be called. Serve the built-in roots without
+ # publishing, so a failure that turns out to be transient is retried
+ # on the next resolution rather than fixed in place.
+ return _builtin_registry()
+ rebuilt = _ResolvedRegistry(
+ provider=provider,
+ registry=validate_unique_root_policies(
+ (*_builtin_purge_policies(), *host_policies)
+ ),
+ # Host roots were validated as they were admitted. Built-in roots stay
+ # lazy, so one unusable built-in cannot block the others.
+ validated={policy.model for policy in host_policies},
+ )
+ globals()["_RESOLVED"] = rebuilt
Review Comment:
**major**: `globals()["_RESOLVED"] = rebuilt` is published with no lock, so
two threads can race. The slower one overwrites the newer snapshot, and the
`validated` set is mutated in `get_purge_policy` without synchronization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
