mikebridge commented on code in PR #41549: URL: https://github.com/apache/superset/pull/41549#discussion_r3658919121
########## superset/commands/deletion_retention/audit.py: ########## @@ -0,0 +1,231 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Write-ahead purge audit record. + +Every purge — time-based or force — writes an immutable record that +**survives** the entity it names, on a **dedicated session** outside the +purge transaction so it neither entangles with the ``DBEventLogger`` +(which shares ``db.session`` and commits mid-request) nor vanishes if the +purge rolls back. The record is written ``pending`` *before* the purge and +flipped to ``confirmed`` *after* it commits, so a crash leaves at most a +``pending`` row, never a missing one. ``pending`` rows are reconciled on the +next run (the purge is convergent). + +The dedicated ``purge_audit_log`` table is content-free (no name or PII; only +action, actor, UTC time, entity type, UUID, and affected referrers) and is never +removed by the purge cascade. +""" + +# Explicit commit/rollback on the dedicated session is the whole point of +# this module — the audit row must survive independently of the purge +# transaction, which the @transaction decorator (scoped to db.session) +# cannot express. +# pylint: disable=consider-using-transaction + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta +from typing import Any, cast +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from sqlalchemy import Column, DateTime, Integer, String, Text +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy_utils import UUIDType + +from superset import db + +logger: logging.Logger = logging.getLogger(__name__) + + +def _dedicated_session() -> Session: + """A fresh session on its own connection, independent of the request / + task ``db.session``. The audit write must commit on its own so it survives + a rolled-back or crashed purge.""" + return sessionmaker(bind=db.engine)() + + +STATUS_PENDING = "pending" +STATUS_CONFIRMED = "confirmed" +STATUS_FAILED = "failed" +STATUS_BLOCKED = "blocked" + +_PENDING_STALE_AFTER = timedelta(hours=1) + +TRIGGER_RETENTION = "retention" +TRIGGER_FORCE = "force" + +ACTOR_SYSTEM = "system" + + +class PurgeAuditLog(db.Model): + """Immutable, content-free record of a purge.""" + + __tablename__ = "purge_audit_log" + + id = Column(UUIDType(binary=True), primary_key=True, default=uuid4) + status = Column(String(16), nullable=False, default=STATUS_PENDING) + trigger = Column(String(16), nullable=False) + actor = Column(String(256), nullable=False) + entity_type = Column(String(64), nullable=False) + entity_uuid = Column(String(36), nullable=True, index=True) + # Comma-joined UUIDs of charts left dangling / dashboards that lost a join + # row (force-purge visibility). Free text, content-free. + affected_referrers = Column(Text, nullable=True) + removed_dashboard_slices = Column(Integer, nullable=False, default=0) + created_on = Column(DateTime, nullable=False) + confirmed_on = Column(DateTime, nullable=True) + + +def write_ahead( + *, + trigger: str, + actor: str, + entity_type: str, + entity_uuid: str | None, + removed_dashboard_slices: int = 0, +) -> UUID | None: + """Insert a ``pending`` audit row on a dedicated session, before the + purge runs. Returns the row id to confirm later, or ``None`` if the audit + write itself fails (which must not block the purge).""" + session = _dedicated_session() + try: + record = PurgeAuditLog( + status=STATUS_PENDING, + trigger=trigger, + actor=actor, + entity_type=entity_type, + entity_uuid=entity_uuid, + removed_dashboard_slices=removed_dashboard_slices, + created_on=datetime.utcnow(), + ) + session.add(record) + session.commit() + return cast(UUID, record.id) + except Exception: # pylint: disable=broad-except + session.rollback() + logger.warning( + "deletion_retention: failed to write pending audit row", exc_info=True + ) + return None + finally: + session.close() + + +def finalize(record_id: UUID | None, status: str, **details: Any) -> None: + """Finalize a pending attempt on the dedicated audit session.""" + if record_id is None: + return + session = _dedicated_session() + try: + record = session.get(PurgeAuditLog, record_id) + if record is None: + return + record.status = status + if status == STATUS_CONFIRMED: + record.confirmed_on = datetime.utcnow() + referrers = details.get("affected_referrers") + if referrers: + record.affected_referrers = ",".join(referrers) + removed_dashboard_slices = details.get("removed_dashboard_slices") + if removed_dashboard_slices is not None: + record.removed_dashboard_slices = removed_dashboard_slices + session.commit() + except Exception: # pylint: disable=broad-except + session.rollback() + logger.warning( + "deletion_retention: failed to finalize audit row %s as %s", + record_id, + status, + exc_info=True, + ) + finally: + session.close() + + +def confirm(record_id: UUID | None, **details: Any) -> None: + """Mark an attempt confirmed after the entity transaction commits.""" + finalize(record_id, STATUS_CONFIRMED, **details) + + +def fail(record_id: UUID | None) -> None: + """Mark a known failed/no-op attempt so it does not remain pending.""" + finalize(record_id, STATUS_FAILED) + + +def block(record_id: UUID | None) -> None: + """Mark an attempt blocked by ordinary deletion policy.""" + finalize(record_id, STATUS_BLOCKED) + + +def _entity_exists(session: Session, record: PurgeAuditLog) -> bool | None: + """Return whether the audit target exists, or None if it cannot resolve.""" + # pylint: disable=import-outside-toplevel + from superset.models.helpers import SoftDeleteMixin + + if record.entity_uuid is None: + return None + for model in SoftDeleteMixin._registered_subclasses: # noqa: SLF001 + table = cast(Any, model).__table__ + if table.name != record.entity_type or "uuid" not in table.c: + continue + return ( + session.execute( + sa.select(table.c.id).where(table.c.uuid == record.entity_uuid).limit(1) + ).first() + is not None + ) + return None + + +def reconcile_pending(stale_before: datetime | None = None) -> dict[str, int]: + """Finalize stale pending attempts left by a process crash. + + Missing entities prove the entity transaction committed, so the attempt is + confirmed. A surviving or unresolvable entity means the attempt did not + durably purge it and is finalized as failed; normal selection may retry. + """ + cutoff = stale_before or datetime.utcnow() - _PENDING_STALE_AFTER + reconciled = confirmed = failed = 0 + session = _dedicated_session() + try: + records = ( + session.query(PurgeAuditLog) + .filter(PurgeAuditLog.status == STATUS_PENDING) + .filter(PurgeAuditLog.created_on < cutoff) + .all() + ) + for record in records: + if _entity_exists(session, record) is False: + record.status = STATUS_CONFIRMED + record.confirmed_on = datetime.utcnow() + confirmed += 1 + else: + record.status = STATUS_FAILED Review Comment: Fixed in 91e9a6930a. (via the shared `_utc_now()` helper). ########## superset/commands/deletion_retention/audit.py: ########## @@ -0,0 +1,231 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Write-ahead purge audit record. + +Every purge — time-based or force — writes an immutable record that +**survives** the entity it names, on a **dedicated session** outside the +purge transaction so it neither entangles with the ``DBEventLogger`` +(which shares ``db.session`` and commits mid-request) nor vanishes if the +purge rolls back. The record is written ``pending`` *before* the purge and +flipped to ``confirmed`` *after* it commits, so a crash leaves at most a +``pending`` row, never a missing one. ``pending`` rows are reconciled on the +next run (the purge is convergent). + +The dedicated ``purge_audit_log`` table is content-free (no name or PII; only +action, actor, UTC time, entity type, UUID, and affected referrers) and is never +removed by the purge cascade. +""" + +# Explicit commit/rollback on the dedicated session is the whole point of +# this module — the audit row must survive independently of the purge +# transaction, which the @transaction decorator (scoped to db.session) +# cannot express. +# pylint: disable=consider-using-transaction + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta +from typing import Any, cast +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from sqlalchemy import Column, DateTime, Integer, String, Text +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy_utils import UUIDType + +from superset import db + +logger: logging.Logger = logging.getLogger(__name__) + + +def _dedicated_session() -> Session: + """A fresh session on its own connection, independent of the request / + task ``db.session``. The audit write must commit on its own so it survives + a rolled-back or crashed purge.""" + return sessionmaker(bind=db.engine)() + + +STATUS_PENDING = "pending" +STATUS_CONFIRMED = "confirmed" +STATUS_FAILED = "failed" +STATUS_BLOCKED = "blocked" + +_PENDING_STALE_AFTER = timedelta(hours=1) + +TRIGGER_RETENTION = "retention" +TRIGGER_FORCE = "force" + +ACTOR_SYSTEM = "system" + + +class PurgeAuditLog(db.Model): + """Immutable, content-free record of a purge.""" + + __tablename__ = "purge_audit_log" + + id = Column(UUIDType(binary=True), primary_key=True, default=uuid4) + status = Column(String(16), nullable=False, default=STATUS_PENDING) + trigger = Column(String(16), nullable=False) + actor = Column(String(256), nullable=False) + entity_type = Column(String(64), nullable=False) + entity_uuid = Column(String(36), nullable=True, index=True) + # Comma-joined UUIDs of charts left dangling / dashboards that lost a join + # row (force-purge visibility). Free text, content-free. + affected_referrers = Column(Text, nullable=True) + removed_dashboard_slices = Column(Integer, nullable=False, default=0) + created_on = Column(DateTime, nullable=False) + confirmed_on = Column(DateTime, nullable=True) + + +def write_ahead( + *, + trigger: str, + actor: str, + entity_type: str, + entity_uuid: str | None, + removed_dashboard_slices: int = 0, +) -> UUID | None: + """Insert a ``pending`` audit row on a dedicated session, before the + purge runs. Returns the row id to confirm later, or ``None`` if the audit + write itself fails (which must not block the purge).""" + session = _dedicated_session() + try: + record = PurgeAuditLog( + status=STATUS_PENDING, + trigger=trigger, + actor=actor, + entity_type=entity_type, + entity_uuid=entity_uuid, + removed_dashboard_slices=removed_dashboard_slices, + created_on=datetime.utcnow(), + ) + session.add(record) + session.commit() + return cast(UUID, record.id) + except Exception: # pylint: disable=broad-except + session.rollback() + logger.warning( + "deletion_retention: failed to write pending audit row", exc_info=True + ) + return None + finally: + session.close() + + +def finalize(record_id: UUID | None, status: str, **details: Any) -> None: + """Finalize a pending attempt on the dedicated audit session.""" + if record_id is None: + return + session = _dedicated_session() + try: + record = session.get(PurgeAuditLog, record_id) + if record is None: + return + record.status = status + if status == STATUS_CONFIRMED: + record.confirmed_on = datetime.utcnow() + referrers = details.get("affected_referrers") + if referrers: + record.affected_referrers = ",".join(referrers) + removed_dashboard_slices = details.get("removed_dashboard_slices") + if removed_dashboard_slices is not None: + record.removed_dashboard_slices = removed_dashboard_slices + session.commit() + except Exception: # pylint: disable=broad-except + session.rollback() + logger.warning( + "deletion_retention: failed to finalize audit row %s as %s", + record_id, + status, + exc_info=True, + ) + finally: + session.close() + + +def confirm(record_id: UUID | None, **details: Any) -> None: + """Mark an attempt confirmed after the entity transaction commits.""" + finalize(record_id, STATUS_CONFIRMED, **details) + + +def fail(record_id: UUID | None) -> None: + """Mark a known failed/no-op attempt so it does not remain pending.""" + finalize(record_id, STATUS_FAILED) + + +def block(record_id: UUID | None) -> None: + """Mark an attempt blocked by ordinary deletion policy.""" + finalize(record_id, STATUS_BLOCKED) + + +def _entity_exists(session: Session, record: PurgeAuditLog) -> bool | None: + """Return whether the audit target exists, or None if it cannot resolve.""" + # pylint: disable=import-outside-toplevel + from superset.models.helpers import SoftDeleteMixin + + if record.entity_uuid is None: + return None + for model in SoftDeleteMixin._registered_subclasses: # noqa: SLF001 + table = cast(Any, model).__table__ + if table.name != record.entity_type or "uuid" not in table.c: + continue + return ( + session.execute( + sa.select(table.c.id).where(table.c.uuid == record.entity_uuid).limit(1) + ).first() + is not None + ) + return None + + +def reconcile_pending(stale_before: datetime | None = None) -> dict[str, int]: + """Finalize stale pending attempts left by a process crash. + + Missing entities prove the entity transaction committed, so the attempt is + confirmed. A surviving or unresolvable entity means the attempt did not + durably purge it and is finalized as failed; normal selection may retry. + """ + cutoff = stale_before or datetime.utcnow() - _PENDING_STALE_AFTER Review Comment: Fixed in 91e9a6930a. (via the shared `_utc_now()` helper). ########## superset/tasks/deletion_retention.py: ########## @@ -0,0 +1,258 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Celery beat task: purge soft-deleted entities past the retention window. + +The deletion-domain analog of ``version_history.prune_old_versions``: where +that ages out version rows while keeping the live entity, this removes +entities that are already soft-deleted. For each +``SoftDeleteMixin`` model it selects rows whose ``deleted_at`` is older than +the per-workspace window and runs the shared cascade per entity, in bounded +id-ordered batches. Convergent, not strictly idempotent: a re-run with the +same clock and data removes nothing, but rows that have since crossed the +cutoff are purged on a later run. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from datetime import datetime, timedelta +from typing import Any, cast + +import sqlalchemy as sa +from flask import current_app + +from superset import db +from superset.commands.deletion_retention import audit +from superset.commands.deletion_retention.purge_cascade import ( + cascade_hard_delete, + CascadeResult, + dashboard_slice_count, + entity_uuid, + suppress_purge_association_versions, +) +from superset.commands.deletion_retention.window import resolve_retention_window +from superset.extensions import celery_app, feature_flag_manager, stats_logger_manager +from superset.models.helpers import ( + skip_visibility_filter, + SoftDeleteMixin, +) + +logger: logging.Logger = logging.getLogger(__name__) + +_METRIC_PREFIX: str = "deletion_retention" +# Below SQLite's historical 999 bind-variable limit; well under PostgreSQL and +# MySQL limits. There is no existing SQLITE_MAX_VARIABLE_NUMBER symbol to reuse. +_PURGE_DELETE_CHUNK: int = 500 +_BATCH: int = _PURGE_DELETE_CHUNK + + +def _soft_delete_models() -> list[type[SoftDeleteMixin]]: + """The registered ``SoftDeleteMixin`` subclasses (dashboards, charts, + datasets), in a stable order.""" + return list(SoftDeleteMixin._registered_subclasses) # noqa: SLF001 + + +def _model_table(model: type[SoftDeleteMixin]) -> sa.Table: + """Return SQLAlchemy table metadata for a registered soft-delete model.""" + return cast(sa.Table, cast(Any, model).__table__) + + +def _model_table_name(model: type[SoftDeleteMixin]) -> str: + """Return the table name for a registered soft-delete model.""" + return str(cast(Any, model).__tablename__) + + +def _iter_eligible_ids( + model: type[SoftDeleteMixin], cutoff: datetime, batch: int +) -> Iterator[list[int]]: + """Yield id-ordered batches of eligible row ids — ``deleted_at IS NOT NULL + AND deleted_at < cutoff`` — querying with the visibility-filter bypass so + soft-deleted rows are visible. Windowed by an ``id`` watermark so memory + and lock-hold stay bounded on a large first run.""" + table = _model_table(model) + after_id = 0 + while True: + with skip_visibility_filter(db.session, model): + ids = [ + row[0] + for row in db.session.execute( + sa.select(table.c.id) + .where(table.c.deleted_at.is_not(None)) + .where(table.c.deleted_at < cutoff) + .where(table.c.id > after_id) + .order_by(table.c.id) + .limit(batch) + ) + ] + if not ids: + return + yield ids + if len(ids) < batch: + return + after_id = ids[-1] + + +def _purge_impl(window_days: int, dry_run: bool) -> dict[str, Any]: + """Run one purge pass across all soft-delete models.""" + if window_days <= 0: + logger.info("deletion_retention: window is 0 (disabled); skipping") + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped") + return {"skipped": 1} + + cutoff = datetime.now() - timedelta(days=window_days) Review Comment: Fixed in 91e9a6930a. The cutoff now uses naive UTC rather than local wall clock. ########## superset/commands/deletion_retention/purge_cascade.py: ########## @@ -0,0 +1,534 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Shared hard-delete cascade for the purge task and force-purge command. + +A single code path keeps the two surfaces from drifting. Every dependent row +is removed by an explicit ``sa.delete``; +the DB ``ON DELETE CASCADE`` constraints are a backstop only — SQLite does +not enforce FKs unless ``PRAGMA foreign_keys=ON`` and Core bulk DML fires +only DB-level cascades, so relying on cascade silently leaks rows. + +Cascade tiers: + +* **M:N join rows** — hard-deleted for the purged entity, including join + rows owned by *surviving* entities (e.g. a live dashboard's + ``dashboard_slices`` row to a purged chart). The entity on the other side + is never touched except to lose that one relationship row. +* **Owned children** (``delete-orphan``, no independent existence) — a + dataset's columns and metrics, hard-deleted with it. +* **Independently-owned entities** (a dashboard's charts, a chart's dataset) + are **preserved**. A live chart's loose ``datasource_id`` to a purged + dataset is left dangling — legacy hard-delete semantics, no guard. +* **Version history** — the entity's own ``*_version`` shadows and + the ``version_changes`` scoped to them, plus an orphan-sweep of any + ``version_transaction`` left owning zero surviving shadows. Runs + behind a ``has_table`` check so it no-ops when versioning is absent. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +logger: logging.Logger = logging.getLogger(__name__) + + +@contextmanager +def suppress_purge_association_versions(session: Session) -> Iterator[None]: + """Discard only association versions generated by this purge session. + + ``dashboard_slices`` is a Continuum-tracked association table. Its + engine-level listener expects a unit-of-work even for Core deletes and + queues a shadow-table insert for each removed relationship. A purge + removes history rather than creating it, so preserve any pending work + that predates this block and discard only statements queued by the purge. + + This is deliberately session-scoped. Mutating Continuum's process-global + ``options["versioning"]`` would let concurrent requests silently lose + unrelated history while a purge is running. + """ + try: + from sqlalchemy_continuum import versioning_manager + except ImportError: + yield + return + + options = versioning_manager.options + if not (options.get("versioning") or options.get("native_versioning")): + yield + return + + unit_of_work = versioning_manager.unit_of_work(session) + pending_before = len(unit_of_work.pending_statements) + try: + yield + finally: + del unit_of_work.pending_statements[pending_before:] + + +def entity_uuid(entity: Any) -> str | None: + """Return the entity's UUID as a string, or ``None`` if it has none.""" + value = getattr(entity, "uuid", None) + return str(value) if value is not None else None + + +def _version_tables_present(bind: Any) -> bool: + """Whether the version-history tables exist (a testable seam, so the + version cascade no-ops cleanly before versioning is installed).""" + return sa.inspect(bind).has_table("version_transaction") + + +@dataclass +class CascadeResult: + """Outcome of one entity's cascade. + + ``purged`` is False when the conditional entity-row delete matched no + row — the entity was restored between selection and delete or was already + gone. In that case + no dependents are touched. + """ + + purged: bool + entity_type: str + entity_uuid: str | None + dangling_chart_uuids: list[str] = field(default_factory=list) + removed_dashboard_slices: int = 0 + version_rows_removed: int = 0 + blocked_reason: str | None = None + + +class PurgeBlockedError(Exception): + """Raised when ordinary deletion policy forbids purging an entity.""" + + +class PurgeRaceLostError(Exception): + """Raised to roll back dependent cleanup when the entity delete loses.""" + + +def cascade_hard_delete( + session: Session, + entity: Any, + *, + enforce_window: bool, + cutoff: datetime | None = None, +) -> CascadeResult: + """Remove *entity* and everything that depends on it in one transaction. + + The entity row is locked and its eligibility is re-checked before any + dependent state is touched. Ordinary deletion blockers remain + authoritative. All cleanup and the conditional parent delete run inside a + savepoint so a lost race or restrictive foreign key leaves no side effects. + """ + # pylint: disable=import-outside-toplevel + from superset.connectors.sqla.models import SqlaTable + from superset.models.slice import Slice + + if enforce_window and cutoff is None: + raise ValueError("cutoff is required when enforce_window=True") + + model = type(entity) + table = model.__table__ + entity_id = entity.id + uuid = entity_uuid(entity) + entity_type = _USER_FACING_TYPE.get(table.name, table.name) + + dangling_chart_uuids: list[str] = [] + removed_dashboard_slices = 0 + version_rows = 0 + permission_name = _dataset_permission_name(entity) if model is SqlaTable else None + + try: + with session.begin_nested(): + claim = sa.select(table.c.id).where(table.c.id == entity_id) + if enforce_window: + claim = claim.where(table.c.deleted_at.is_not(None)).where( + table.c.deleted_at < cutoff + ) + if session.execute(claim.with_for_update()).scalar_one_or_none() is None: + raise PurgeRaceLostError + + _validate_deletion_allowed(session, model, entity_id) + removed_dashboard_slices = _count_dashboard_slices( + session, model, entity_id + ) + if model is SqlaTable: + dangling_chart_uuids = [ + str(chart_uuid) + for (chart_uuid,) in session.execute( + sa.select(Slice.uuid) + .where(Slice.datasource_id == entity_id) + .where(Slice.datasource_type == "table") + ) + ] + + _delete_m2m_joins(session, model, entity_id) + _delete_owned_children(session, model, entity_id) + version_rows = _delete_version_history(session, entity, entity_id) Review Comment: Fixed in 91e9a6930a. `version_rows_removed` now includes the version_changes delete rowcount. ########## superset/commands/deletion_retention/audit.py: ########## @@ -0,0 +1,225 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Write-ahead purge audit record. + +Every purge — time-based or force — writes an immutable record that +**survives** the entity it names, on a **dedicated session** outside the +purge transaction so it neither entangles with the ``DBEventLogger`` +(which shares ``db.session`` and commits mid-request) nor vanishes if the +purge rolls back. The record is written ``pending`` *before* the purge and +flipped to ``confirmed`` *after* it commits, so a crash leaves at most a +``pending`` row, never a missing one. ``pending`` rows are reconciled on the +next run (the purge is convergent). + +The dedicated ``purge_audit_log`` table is content-free (no name or PII; only +action, actor, UTC time, entity type, UUID, and affected referrers) and is never +removed by the purge cascade. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta +from typing import Any, cast +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from sqlalchemy import Column, DateTime, Integer, String, Text +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy_utils import UUIDType + +from superset import db + +logger: logging.Logger = logging.getLogger(__name__) + + +def _dedicated_session() -> Session: + """A fresh session on its own connection, independent of the request / + task ``db.session``. The audit write must commit on its own so it survives + a rolled-back or crashed purge.""" + return sessionmaker(bind=db.engine)() + + +STATUS_PENDING = "pending" +STATUS_CONFIRMED = "confirmed" +STATUS_FAILED = "failed" +STATUS_BLOCKED = "blocked" + +_PENDING_STALE_AFTER = timedelta(hours=1) + +TRIGGER_RETENTION = "retention" +TRIGGER_FORCE = "force" + +ACTOR_SYSTEM = "system" + + +class PurgeAuditLog(db.Model): + """Immutable, content-free record of a purge.""" + + __tablename__ = "purge_audit_log" + + id = Column(UUIDType(binary=True), primary_key=True, default=uuid4) + status = Column(String(16), nullable=False, default=STATUS_PENDING) + trigger = Column(String(16), nullable=False) + actor = Column(String(256), nullable=False) + entity_type = Column(String(64), nullable=False) + entity_uuid = Column(String(36), nullable=True, index=True) + # Comma-joined UUIDs of charts left dangling / dashboards that lost a join + # row (force-purge visibility). Free text, content-free. + affected_referrers = Column(Text, nullable=True) + removed_dashboard_slices = Column(Integer, nullable=False, default=0) + created_on = Column(DateTime, nullable=False) + confirmed_on = Column(DateTime, nullable=True) + + +def write_ahead( + *, + trigger: str, + actor: str, + entity_type: str, + entity_uuid: str | None, + removed_dashboard_slices: int = 0, +) -> UUID | None: + """Insert a ``pending`` audit row on a dedicated session, before the + purge runs. Returns the row id to confirm later, or ``None`` if the audit + write itself fails (which must not block the purge).""" + session = _dedicated_session() + try: + record = PurgeAuditLog( + status=STATUS_PENDING, + trigger=trigger, + actor=actor, + entity_type=entity_type, + entity_uuid=entity_uuid, + removed_dashboard_slices=removed_dashboard_slices, + created_on=datetime.utcnow(), + ) + session.add(record) + session.commit() + return cast(UUID, record.id) + except Exception: # pylint: disable=broad-except + session.rollback() + logger.warning( + "deletion_retention: failed to write pending audit row", exc_info=True + ) + return None + finally: + session.close() + + +def finalize(record_id: UUID | None, status: str, **details: Any) -> None: + """Finalize a pending attempt on the dedicated audit session.""" + if record_id is None: + return + session = _dedicated_session() + try: + record = session.get(PurgeAuditLog, record_id) + if record is None: + return + record.status = status + if status == STATUS_CONFIRMED: + record.confirmed_on = datetime.utcnow() Review Comment: Fixed in 91e9a6930a. `finalize()` only transitions rows still pending, so a delayed worker cannot overwrite a reconciled outcome. ########## tests/integration_tests/deletion_retention/purge_tests.py: ########## @@ -0,0 +1,447 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Integration coverage for the time-based soft-delete purge. + +Exercises ``superset.tasks.deletion_retention`` against a real database: +the cascade (M:N joins, owned children, datasource permission, version +shadows), preservation of surviving entities, dry-run, the explicit-delete +guarantee under FK enforcement OFF, and the version-tables-absent no-op. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import Any +from unittest.mock import MagicMock, patch + +import sqlalchemy as sa +from sqlalchemy.sql.dml import Delete + +from superset import db, security_manager +from superset.commands.deletion_retention import audit +from superset.commands.deletion_retention.purge_cascade import cascade_hard_delete +from superset.connectors.sqla.models import ( + RLSFilterTables, + RowLevelSecurityFilter, + SqlaTable, +) +from superset.models.dashboard import Dashboard +from superset.models.slice import Slice +from superset.models.user_attributes import UserAttribute +from superset.reports.models import ReportSchedule +from superset.tags.models import ObjectType, Tag, TaggedObject +from superset.tasks.deletion_retention import _purge_impl + +from ._base import DeletionRetentionTestBase + + +def _purge(window: int = 30, dry_run: bool = False) -> dict[str, Any]: + return _purge_impl(window, dry_run) + + +class TestSoftDeletePurge(DeletionRetentionTestBase): + def test_aged_out_purged_in_window_and_active_preserved(self) -> None: + """Rows past the window are purged; in-window + soft-deleted and active rows are preserved.""" + aged = self.make_chart("aged") + recent = self.make_chart("recent") + active = self.make_chart("active") + aged_id, recent_id, active_id = aged.id, recent.id, active.id + self.soft_delete(aged, days_ago=90) + self.soft_delete(recent, days_ago=5) + + result = _purge(window=30) + + assert result["purged"].get("slices") == 1, result + assert not self.exists(Slice, aged_id) + assert self.exists(Slice, recent_id) + assert self.exists(Slice, active_id) + + def test_window_zero_disables(self) -> None: + """Window 0 disables the time-based purge.""" + aged = self.make_chart("aged") + aged_id = aged.id + self.soft_delete(aged, days_ago=90) + + assert _purge(window=0) == {"skipped": 1} + assert self.exists(Slice, aged_id) + + def test_dry_run_removes_nothing(self) -> None: + """Dry-run reports would_purge but deletes nothing.""" + aged = self.make_chart("aged") + aged_id = aged.id + self.soft_delete(aged, days_ago=90) + + result = _purge(window=30, dry_run=True) + + assert result["would_purge"].get("slices") == 1, result + assert self.exists(Slice, aged_id) + + def test_purging_chart_unlinks_live_dashboard_but_keeps_it(self) -> None: + """Purging a chart removes its dashboard_slices + rows — including one on a *live* dashboard — but the dashboard and + other charts survive.""" + chart = self.make_chart("c") + other = self.make_chart("other") + chart_id, other_id = chart.id, other.id + dash = self.make_dashboard("live", slices=[chart, other]) + dash_id = dash.id + self.soft_delete(chart, days_ago=90) + + _purge(window=30) + + assert not self.exists(Slice, chart_id) + assert self.exists(Slice, other_id) + assert self.exists(Dashboard, dash_id) + remaining = self.count( + "SELECT count(*) FROM dashboard_slices WHERE slice_id = :i", + {"i": chart_id}, + ) + assert remaining == 0 + # the live dashboard keeps its link to the surviving chart + assert ( + self.count( + "SELECT count(*) FROM dashboard_slices WHERE dashboard_id = :d", + {"d": dash_id}, + ) + == 1 + ) + + def test_purging_dashboard_preserves_its_charts(self) -> None: + """Purging a dashboard does not remove the + independently-owned charts it referenced.""" + chart = self.make_chart("kept") + chart_id = chart.id + dash = self.make_dashboard("doomed", slices=[chart]) + dash_id = dash.id + self.soft_delete(dash, days_ago=90) + + _purge(window=30) + + assert not self.exists(Dashboard, dash_id) + assert self.exists(Slice, chart_id) + assert ( + self.count( + "SELECT count(*) FROM dashboard_slices WHERE dashboard_id = :d", + {"d": dash_id}, + ) + == 0 + ) + + def test_purging_dataset_removes_children_and_permission(self) -> None: + """A dataset's owned columns, metrics, and datasource + permission are removed with it.""" + dataset = self.make_dataset("withchildren", with_children=True) + ds_id = dataset.id + vm_name = security_manager.get_dataset_perm( + dataset.id, dataset.table_name, dataset.database.database_name + ) + assert security_manager.find_permission_view_menu( + "datasource_access", vm_name + ), "fixture should have created the datasource PVM" + self.soft_delete(dataset, days_ago=90) + + _purge(window=30) + + assert not self.exists(SqlaTable, ds_id) + assert ( + self.count( + "SELECT count(*) FROM table_columns WHERE table_id = :i", {"i": ds_id} + ) + == 0 + ) + assert ( + self.count( + "SELECT count(*) FROM sql_metrics WHERE table_id = :i", {"i": ds_id} + ) + == 0 + ) + assert not security_manager.find_permission_view_menu( + "datasource_access", vm_name + ) + + def test_restore_race_does_not_remove_dataset_permission(self) -> None: + """A zero-row conditional parent delete leaves its permission intact.""" + dataset = self.make_dataset("restore_race") + vm_name = security_manager.get_dataset_perm( + dataset.id, dataset.table_name, dataset.database.database_name + ) + self.soft_delete(dataset, days_ago=90) + session = db.session() + execute = session.execute + + def lose_parent_delete(statement: Any, *args: Any, **kwargs: Any) -> Any: + if isinstance(statement, Delete) and statement.table.name == "tables": + return MagicMock(rowcount=0) + return execute(statement, *args, **kwargs) + + with patch.object(session, "execute", side_effect=lose_parent_delete): + result = cascade_hard_delete( + session, + dataset, + enforce_window=True, + cutoff=datetime.now() - timedelta(days=30), + ) + session.commit() + + assert result.purged is False + assert self.exists(SqlaTable, dataset.id) + assert security_manager.find_permission_view_menu("datasource_access", vm_name) + + def test_dataset_purge_removes_rls_join_but_preserves_rule(self) -> None: + """RLS M:N rows follow ordinary ORM cleanup; the rule survives.""" + dataset = self.make_dataset("rls") + rule = RowLevelSecurityFilter( + name="retention_it_rls", + clause="1=1", + filter_type="Regular", + tables=[dataset], + ) + db.session.add(rule) + db.session.commit() + rule_id, dataset_id = rule.id, dataset.id + self.soft_delete(dataset, days_ago=90) + + result = _purge(window=30) + + assert result["purged"].get("tables") == 1 + assert db.session.get(RowLevelSecurityFilter, rule_id) is not None + assert ( + db.session.execute( + sa.select(sa.func.count()) + .select_from(RLSFilterTables) + .where(RLSFilterTables.c.table_id == dataset_id) + ).scalar_one() + == 0 + ) + + def test_report_reference_blocks_chart_purge(self) -> None: + """Retention preserves the ordinary chart/report deletion guard.""" + chart = self.make_chart("reported") + report = ReportSchedule( + type="Report", + name="retention_it_report", + crontab="0 0 * * *", + chart=chart, + ) + db.session.add(report) + db.session.commit() + chart_id, chart_uuid = chart.id, str(chart.uuid) + self.soft_delete(chart, days_ago=90) + + result = _purge(window=30) + + assert result["blocked_by_reference"] == 1 + assert result["purged"].get("slices", 0) == 0 + assert self.exists(Slice, chart_id) + assert db.session.get(ReportSchedule, report.id) is not None + row = ( + db.session.query(audit.PurgeAuditLog) + .filter_by(entity_uuid=chart_uuid) + .one() + ) + assert row.status == audit.STATUS_BLOCKED + + def test_restrictive_fk_blocks_dashboard_without_rewriting_referrer(self) -> None: + """A welcome-dashboard FK remains authoritative during retention.""" + dashboard = self.make_dashboard("welcome") + dashboard_id = dashboard.id + user = self.get_user("admin") + attribute = ( + db.session.query(UserAttribute).filter_by(user_id=user.id).one_or_none() + ) + created = attribute is None + if attribute is None: + attribute = UserAttribute(user_id=user.id) + db.session.add(attribute) + previous_dashboard_id = attribute.welcome_dashboard_id + attribute.welcome_dashboard_id = dashboard_id + db.session.commit() + self.soft_delete(dashboard, days_ago=90) + + try: + result = _purge(window=30) + + assert result["blocked_by_reference"] == 1 + assert self.exists(Dashboard, dashboard_id) + db.session.refresh(attribute) + assert attribute.welcome_dashboard_id == dashboard_id + finally: + if created: + db.session.delete(attribute) + else: + attribute.welcome_dashboard_id = previous_dashboard_id + db.session.commit() + + def test_purging_dataset_leaves_referencing_chart_dangling(self) -> None: + """A soft-deleted dataset is purged without a dependent guard even + with a live chart referencing it; the chart is left + dangling (unchanged), not blocked or rewritten.""" + chart = self.make_chart("dangling", dataset=self.dataset) + chart_id, ds_id = chart.id, self.dataset.id + self.soft_delete(self.dataset, days_ago=90) + + _purge(window=30) + + assert not self.exists(SqlaTable, ds_id) + assert self.exists(Slice, chart_id) + kept = db.session.query(Slice).filter(Slice.id == chart_id).one() + assert kept.datasource_id == ds_id # dangling, unmodified + + def test_tags_removed_on_purge(self) -> None: + """The entity's tagged_object rows are removed (the + after_delete tag-cleanup Core bulk-delete skips).""" + chart = self.make_chart("tagged") + chart_id = chart.id + tag = Tag(name="retention_it_tag") + db.session.add(tag) + db.session.commit() + db.session.add( + TaggedObject( + tag_id=tag.id, object_id=chart_id, object_type=ObjectType.chart + ) + ) + db.session.commit() + self.soft_delete(chart, days_ago=90) + + _purge(window=30) + + assert ( + self.count( + "SELECT count(*) FROM tagged_object WHERE object_id = :i " + "AND object_type = 'chart'", + {"i": chart_id}, + ) + == 0 + ) + + def test_soft_delete_and_restore_create_no_version_rows(self) -> None: + """deleted_at is excluded from versioning: soft-delete and restore are + state changes, not edits, so (with capture ON) they add no version + rows. Versioning and deletion are orthogonal — restore is a separate + state flip, never a version-shadow operation.""" + chart = self.make_chart("orthogonal") + cid = chart.id + baseline = self.count( + "SELECT count(*) FROM slices_version WHERE id = :i", {"i": cid} + ) + + chart.soft_delete() # state change, not an edit + db.session.commit() + after_delete = self.count( + "SELECT count(*) FROM slices_version WHERE id = :i", {"i": cid} + ) + + chart.restore() # state flip back, not an edit + db.session.commit() + after_restore = self.count( + "SELECT count(*) FROM slices_version WHERE id = :i", {"i": cid} + ) + + assert after_delete == baseline, "soft-delete must not create a version row" + assert after_restore == baseline, "restore must not create a version row" + + def test_cascade_with_fk_enforcement_off(self) -> None: + """The explicit sa.delete cascade, not the database FK cascade, + does the work. With SQLite FK enforcement OFF, no junction rows are + orphaned.""" + if db.engine.dialect.name != "sqlite": + self.skipTest("FK-off probe is SQLite-specific") + chart = self.make_chart("fkoff") + chart_id = chart.id + self.make_dashboard("fkoffdash", slices=[chart]) + self.soft_delete(chart, days_ago=90) + + db.session.execute(sa.text("PRAGMA foreign_keys=OFF")) + _purge(window=30) + + assert not self.exists(Slice, chart_id) Review Comment: Fixed in 91e9a6930a. The probe restores `PRAGMA foreign_keys=ON` in a finally block. -- 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]
