sadpandajoe commented on code in PR #44892:
URL: https://github.com/apache/superset/pull/44892#discussion_r4206852292
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1414,308 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ The shared cleanup empties associations before owned children, while an
+ owned table's predicate selects its rows *through* its ownership path. If
+ a hop on that path is an association, its rows are already gone when the
+ owned delete runs, so the statement matches nothing.
+
+ Kept where the rest of the ordering rules were dropped because this one
+ can be silent: where foreign keys are enforced the association delete
+ fails loudly, but where they are not -- SQLite -- the root is purged and
+ its descendants are left orphaned with nothing reported. The remedy is to
+ classify the intermediate table as owned, which places it in the same
+ phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
- raise RuntimeError(
- f"Missing listener action for {dependency.key.describe()}"
- )
- if (
- dependency.classification is DependencyClassification.VERSION_OWNED
- and dependency.version_column is None
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage, and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _validate_recursive_ownership(policy: PurgeEntityPolicy) -> None:
+ """Reject a self-referencing owned table under one-level cleanup.
+
+ Cleanup that issues one statement per declared edge prunes a table that
+ owns itself one level deep. Where foreign keys are enforced the root's own
+ delete then fails and rolls back; where they are not -- SQLite -- the root
+ is purged and its grandchildren are left behind with a dangling parent id
+ and nothing reported. A policy declaring a tree walks it itself and says
+ so with ``walks_own_subtree``.
+ """
+ for dependency in policy.dependencies:
+ if dependency.classification is not DependencyClassification.OWNED:
Review Comment:
A host root with a `parent_id` self-FK that classifies the discovered edge
as `ASSOCIATION` skips this guard (only `OWNED` is refused here), so the stock
association cleanup runs `DELETE FROM host_node WHERE parent_id IN (SELECT id
FROM host_node WHERE id = :root)`. That hard-deletes the live, non-archived
direct children of the purged row with no eligibility predicate or audit row,
and leaves grandchildren with a dangling `parent_id` (on SQLite, or when the
child is a leaf, the purge commits as a clean success). The same applies to an
`ASSOCIATION`/`OWNED` edge whose `related_table` is another purge root's table.
Should admission refuse executable declarations whose related table is the
root's own table or another registered root's table, regardless of
`walks_own_subtree`?
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +931,443 @@ 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, ...]:
+ """Materialize the provider's payload, or reject its shape.
+
+ 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.
+
+ An exception raised *while reading* a lazy payload is left to propagate.
+ Walking a generator runs host code just as calling the provider does, so
+ the two are the same kind of failure and earn the same retry; only the
+ shape of a payload that arrived intact is settled for good.
+ """
+ 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 ()
+ return tuple(provided)
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+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
+ # Version shadows are resolved by bare name as well, in
+ # _entity_version_targets.
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ or dependency.classification is DependencyClassification.VERSION_OWNED
+ }
+ 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_tables: frozenset[str],
+ 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:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ 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
+ root_table: sa.Table = sa.inspect(candidate.model).local_table
+ if root_table.key in builtin_tables:
+ # A different class mapped onto a built-in table -- a subclass of
+ # Slice, say -- is not the built-in root by identity, but purging
+ # it runs host cleanup against this package's own rows. It also
+ # joins the soft-delete registry through the mixin, so the
+ # scheduled task would scan it.
+ logger.error(
+ "purge_policy: host policy root %s is mapped onto built-in "
+ "table %r; ignored",
+ candidate.model.__name__,
+ root_table.key,
+ )
+ return None
+ if not isinstance(candidate.entity_type, str) or not
candidate.entity_type:
+ # Read as a string by the cascade, the audit record and every log
+ # line that names the root.
+ logger.error(
+ "purge_policy: host policy for %s declares entity_type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
Review Comment:
Admission reserves built-in models, tables and `entity_type` values, but the
version cascade still keys `version_changes` cleanup on the bare class name
(`purge_cascade.py` maps `model.__name__` through `ENTITY_KIND_BY_CLASS_NAME`).
A versioned host class that happens to be named `Dashboard` (own table, own
`entity_type`) passes admission, and purging its row 7 then deletes the
`version_changes` rows for the built-in dashboard 7, including a live one.
Should admission also reject a host model whose `__name__` is in
`ENTITY_KIND_BY_CLASS_NAME`, or should the cleanup bind to the policy's model
identity instead?
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1414,308 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ The shared cleanup empties associations before owned children, while an
+ owned table's predicate selects its rows *through* its ownership path. If
+ a hop on that path is an association, its rows are already gone when the
+ owned delete runs, so the statement matches nothing.
+
+ Kept where the rest of the ordering rules were dropped because this one
+ can be silent: where foreign keys are enforced the association delete
+ fails loudly, but where they are not -- SQLite -- the root is purged and
+ its descendants are left orphaned with nothing reported. The remedy is to
+ classify the intermediate table as owned, which places it in the same
+ phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
- raise RuntimeError(
- f"Missing listener action for {dependency.key.describe()}"
- )
- if (
- dependency.classification is DependencyClassification.VERSION_OWNED
- and dependency.version_column is None
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage, and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _validate_recursive_ownership(policy: PurgeEntityPolicy) -> None:
+ """Reject a self-referencing owned table under one-level cleanup.
+
+ Cleanup that issues one statement per declared edge prunes a table that
+ owns itself one level deep. Where foreign keys are enforced the root's own
+ delete then fails and rolls back; where they are not -- SQLite -- the root
+ is purged and its grandchildren are left behind with a dangling parent id
+ and nothing reported. A policy declaring a tree walks it itself and says
+ so with ``walks_own_subtree``.
+ """
+ for dependency in policy.dependencies:
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ key: DependencyKey = dependency.key
+ if key.owner_table != key.related_table:
+ continue
+ raise RuntimeError(
+ f"Owned dependency {key.describe()} is self-referencing; the stock
"
+ "owned-child cleanup deletes one level, so this policy must supply
"
+ "its own delete_owned_children"
+ )
+
+
+def _validate_scanner_requirements(policy: PurgeEntityPolicy) -> None:
+ """Reject a root the scheduled scan cannot page through.
+
+ The retention task selects, windows and orders eligible rows by ``id``,
+ and the cascade pins each row by it. A root keyed on something else -- a
+ UUID primary key with no ``id`` column -- raises inside the scan, outside
+ the per-row error handling, so the run aborts before the remaining roots
+ are reached.
+ """
+ mapper: Mapper[Any] = sa.inspect(policy.model)
+ table: sa.Table = mapper.local_table
+ if "id" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'id' column; the "
+ "scheduled scan pages eligible rows by id"
+ )
+ primary_key: tuple[sa.Column[Any], ...] = tuple(mapper.primary_key)
+ if (
+ len(primary_key) != 1
+ or primary_key[0].name != "id"
+ or not isinstance(primary_key[0].type, sa.Integer)
+ ):
+ # The scan pages by id and the row is then fetched with a scalar
+ # Session.get, so a composite or differently named key matches
+ # nothing: every row fails while a dry run still counts it eligible.
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} is keyed on "
+ f"({', '.join(column.name for column in primary_key)}); the "
+ "scheduled purge pages an integer 'id' watermark and looks rows "
+ "up by it"
+ )
+ if "deleted_at" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'deleted_at' column; "
+ "every purge path requires the row to be archived first"
+ )
+ uuid_column: sa.Column[Any] | None = table.c.get("uuid")
+ if (
+ uuid_column is not None
+ and isinstance(uuid_column.type, sa.Uuid)
+ and not uuid_column.type.native_uuid
+ ):
+ # The third column the frame reads: the locked claim and the
+ # conditional delete both carry ``uuid == str(entity.uuid)``. A
+ # character-based UUID type binds through a processor that expects a
+ # UUID object, so every eligible row would raise and stay archived
+ # while a dry run counted it purgeable -- on every dialect, which is
+ # what makes it checkable here.
+ #
+ # ``native_uuid`` is left alone deliberately: it has no bind processor
+ # where the database has a UUID type of its own, so the string binds
+ # cleanly on Postgres and only a database without one rejects it. That
+ # is a per-row failure like any other, and the docs say so.
+ # ``sqlalchemy_utils.UUIDType`` is not a subclass of this at all and
+ # accepts the string everywhere, which is why the built-in roots are
+ # unaffected.
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} types 'uuid' as a "
+ "character-based sa.Uuid; the locked claim binds that value as a "
+ "string, which the type rejects on every dialect"
+ )
+
+
+#: Classifications the shared cleanup executes as SQL, through
+#: ``_dependency_predicates`` -- which only knows how to read an inbound
+#: foreign key.
+_EXECUTABLE_CLASSIFICATIONS: frozenset[DependencyClassification] = frozenset(
+ {
+ DependencyClassification.OWNED,
+ DependencyClassification.ASSOCIATION,
+ DependencyClassification.BLOCK,
+ }
+)
+
+
+def _validate_versioned_root(policy: PurgeEntityPolicy) -> None:
+ """Reject version targets on a root that has no version class.
+
+ ``_delete_version_history`` resolves the *root's* version class first and
+ returns as soon as that raises, so a target declared on a child's shadow
+ is never reached for an unversioned root: the root and its live child rows
+ are purged while the child's history survives, with nothing reported.
+
+ Skipped where nothing is versioned at all. The version cascade is then
+ inert for every root, including this package's own, and refusing would
+ only take those down with it.
+ """
+ try:
+ from sqlalchemy_continuum import version_class, versioning_manager
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+ except ImportError: # pragma: no cover - versioning not installed
+ return
+ if not getattr(versioning_manager, "version_class_map", None):
+ return
+ try:
+ version_class(policy.model)
+ except ClassNotVersioned:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no version class, so the "
+ "version cascade returns before reaching a declared target; such "
+ "a policy declares none"
+ ) from None
+
+
+def _validate_version_target(
+ policy: PurgeEntityPolicy, dependency: DependencyPolicy, metadata:
sa.MetaData
+) -> None:
+ """Reject a version target whose column does not hold the root's id.
+
+ Version cleanup compares the declared column with the root's id directly,
+ so the column has to be root-relative. Shadow tables are generated and may
+ not exist when a policy is validated, so the check reads their live
+ counterpart -- ``<table>_version`` describes ``<table>``, the naming the
+ shadow declarations already rely on -- and accepts the column when it is
+ the root's own primary key or a foreign key into the root.
+
+ A column naming an intermediate owner, a stage id under a workflow root
+ say, matches history belonging to a different root and leaves this one's
+ behind.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ column: str = cast(str, dependency.version_column)
+ shadow_name: str = dependency.key.related_table
+ suffix: str = "_version"
+ live_name: str = (
+ shadow_name[: -len(suffix)] if shadow_name.endswith(suffix) else
shadow_name
Review Comment:
When the declared target has no `_version` suffix, `live_name` falls back to
the target itself, so a `VERSION_OWNED` dependency naming a live table (say
`notes` with `version_column="project_id"`) clears root-relative validation if
that column is a foreign key to the root. Version cleanup then resolves the
live table and runs `DELETE FROM notes WHERE project_id = ...`, removing rows a
`PRESERVE` declaration meant to keep while the real shadow history is
untouched. Should a target that is not a `*_version` shadow be rejected here?
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +931,443 @@ 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, ...]:
+ """Materialize the provider's payload, or reject its shape.
+
+ 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.
+
+ An exception raised *while reading* a lazy payload is left to propagate.
+ Walking a generator runs host code just as calling the provider does, so
+ the two are the same kind of failure and earn the same retry; only the
+ shape of a payload that arrived intact is settled for good.
+ """
+ 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 ()
+ return tuple(provided)
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+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
+ # Version shadows are resolved by bare name as well, in
+ # _entity_version_targets.
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ or dependency.classification is DependencyClassification.VERSION_OWNED
+ }
+ 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_tables: frozenset[str],
+ 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:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ 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
+ root_table: sa.Table = sa.inspect(candidate.model).local_table
+ if root_table.key in builtin_tables:
+ # A different class mapped onto a built-in table -- a subclass of
+ # Slice, say -- is not the built-in root by identity, but purging
+ # it runs host cleanup against this package's own rows. It also
+ # joins the soft-delete registry through the mixin, so the
+ # scheduled task would scan it.
+ logger.error(
+ "purge_policy: host policy root %s is mapped onto built-in "
+ "table %r; ignored",
+ candidate.model.__name__,
+ root_table.key,
+ )
+ return None
+ if not isinstance(candidate.entity_type, str) or not
candidate.entity_type:
+ # Read as a string by the cascade, the audit record and every log
+ # line that names the root.
+ logger.error(
+ "purge_policy: host policy for %s declares entity_type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ 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:
+ # Two different reasons, both about identities this package owns.
+ # A stock listener action decides what to delete from a built-in
+ # entity type, so declared by a host it is accepted and then never
+ # runs. A blocker *would* be applied -- validate_deletion_allowed
+ # is generic over the declarations -- but its reason code lands in
+ # the purge audit record, whose vocabulary is a closed set pinned
+ # by test. A host refuses a purge from its own validator instead,
+ # where the code it records is plainly its own.
+ raise RuntimeError(
+ f"{', '.join(sorted(unusable))} dependencies are not available
"
+ "to 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 failures that may be transient -- the provider
+ could not be called, or a lazy payload raised while being read. The caller
+ then publishes the built-in roots with a short deadline, so the next
+ resolution after it tries again rather than treating one bad moment as the
+ answer for the life of the process. A payload that arrived intact but is
+ the wrong shape is a host bug that will not fix itself, so it is published
+ as an empty result instead.
+ """
+ if provider is None:
+ return ()
+ if not callable(provider):
+ # Static misconfiguration, not an outage: reported once rather than
+ # retried with a traceback for the life of the process.
+ logger.error(
+ "purge_policy: %s is %s, not a callable returning policies",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provider).__name__,
+ )
+ return ()
+ try:
+ # Both the call and reading what it returns run host code, so a lazy
+ # payload that raises mid-walk is the same kind of failure as a
+ # provider that raises outright, and earns the same retry.
+ candidates: tuple[Any, ...] = _host_policy_payload(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
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_tables: frozenset[str] = frozenset(
+ sa.inspect(policy.model).local_table.key 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_tables, 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.
+
+ Matched to its provider by *identity* rather than keyed in an
+ ``lru_cache``: a cache key would have to be hashed, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That raises
+ before the host boundary can isolate the host's mistake, which would
+ 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)
+ #: Set only on a snapshot published because the provider could not be
+ #: called. Until this deadline the snapshot answers reads; after it, the
+ #: provider is tried again.
+ retry_after: float | None = None
+
+
+#: Key the resolved index lives under on the Flask app. Per app rather than
+#: per process: two apps in one process have different configs, and a single
+#: global snapshot would thrash between them -- re-resolving, and re-invoking
+#: the provider, on every read.
+_REGISTRY_EXTENSION_KEY: str = "deletion_retention_purge_registry"
+
+#: How long a snapshot published after a provider failure answers reads
+#: before the provider is tried again. Bounds both extremes: a failure is
+#: neither remembered for the life of the process nor retried on every read --
+#: at roughly three reads per purged entity, retrying per read meant tens of
+#: thousands of provider calls and tracebacks in a single pass.
+_PROVIDER_RETRY_SECONDS: float = 60.0
+
+#: 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.
+ """
+ if not has_app_context():
+ return _builtin_registry()
+ provider: Callable[[], Any] | None = current_app.config.get(
+ HOST_POLICIES_CONFIG_KEY
+ )
+ published: _ResolvedRegistry | None = current_app.extensions.get(
+ _REGISTRY_EXTENSION_KEY
+ )
+ if published is None:
+ published = _builtin_registry()
+ current_app.extensions[_REGISTRY_EXTENSION_KEY] = published
+ if published.provider is provider and (
+ published.retry_after is None or time.monotonic() <
published.retry_after
+ ):
+ return published
+ if getattr(_RESOLVING, "active", False):
+ # A nested resolution: the provider is building its policy from a
+ # built-in one, so hand it the roots this package declares rather than
+ # re-entering a resolution that has not finished.
+ return _builtin_registry()
+ # Not nested: the branch above returns when the marker is already set,
+ # so reaching here means this thread was not resolving.
+ _RESOLVING.active = True
+ try:
+ host_policies: tuple[PurgeEntityPolicy, ...] | None =
_host_purge_policies(
+ provider
+ )
+ finally:
+ _RESOLVING.active = False
+ if host_policies is None:
+ current: _ResolvedRegistry | None = current_app.extensions.get(
+ _REGISTRY_EXTENSION_KEY
+ )
+ if (
+ current is not None
+ and current.provider is provider
+ and current.retry_after is None
Review Comment:
This branch keeps a good snapshot when a slower caller fails after a faster
one published, but no test pins that ordering: the existing concurrent test has
both calls succeed, and the sequential retry tests never reach it. If the guard
were removed, the late failure would overwrite the host index with built-ins
only and host roots would report as unsupported for the retry window. Could a
registry unit test use events to publish a successful resolution before a
second call's provider raises, then assert both that caller and a later lookup
still see the host policy without another provider call?
--
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]
