diff --git a/UPDATING.md b/UPDATING.md index d8d0fe901d0..7892438f7cb 100644 --- a/UPDATING.md +++ b/UPDATING.md @@ -408,6 +408,16 @@ Entity version history (the `version_transaction` / `*_version` shadow tables th The task ships in the default `CeleryConfig.beat_schedule`; a deployment that overrides `CELERY_CONFIG` without inheriting the default will log a startup warning that the prune task is absent (so it never silently stops running). Retention only prunes whatever history exists — capture itself is gated separately by `ENABLE_VERSIONING_CAPTURE` (ships off). +### Deletion retention (soft-deleted entities are eventually purged) + +Soft-deleted dashboards, charts, and datasets are now permanently removed after a retention window (default 30 days; `SOFT_DELETE_RETENTION_DAYS`, `0` disables; settable per workspace at runtime via the `deletion-retention set-window` CLI, which takes precedence). The `deletion_retention.purge_soft_deleted` Celery beat task runs daily and removes each aged-out entity together with its M:N join rows, owned children, datasource permission, and version-history shadow rows. After purge an entity is **unrecoverable** — its detail and `/restore` endpoints return 404 and its version history is gone. + +The introducing release **defaults to dry-run** (`SOFT_DELETE_PURGE_DRY_RUN=True`): the task logs `would_purge` counts but deletes nothing, so operators can validate against production before activating real purging by setting it to `False`. Note `would_purge` is an **upper bound** — it counts every entity past the retention window without evaluating deletion blockers, so a real run may purge fewer (entities referenced by report schedules or set as a user's welcome dashboard are blocked and reported separately). The task only acts while the temporary `SOFT_DELETE` rollout flag is on. + +Deployments that replace the default `CELERY_CONFIG` must add `superset.tasks.deletion_retention` to the Celery `imports` and schedule the `deletion_retention.purge_soft_deleted` task themselves. The shipped Docker development config includes both entries. + +Operators can immediately erase a specific entity for compliance (GDPR) via `superset deletion-retention force-purge --uuid `; this applies legacy hard-delete semantics — a live chart referencing a force-purged dataset is left without a datasource until re-pointed (the chart is not modified), and it purges the named entity even when it was never soft-deleted. Every purge writes an immutable, content-free audit record to the new `purge_audit_log` table that survives the entity it names: the **scheduled** purge fails closed (an entity whose audit row cannot be written is skipped and retried next run), while **force-purge** proceeds even if the audit write fails — the operator is present and deletion outranks audit for a compliance erasure. + ### Webhook alerts/reports block private/internal hosts by default Webhook alert/report dispatch (`WebhookNotification.send`) now validates the target URL's host against the same private/internal-IP block applied to dataset import URLs. If the resolved host is in a loopback, link-local, private (RFC-1918), shared-CGNAT, or multicast range, the webhook is rejected with `NotificationParamException`. diff --git a/docker/pythonpath_dev/superset_config.py b/docker/pythonpath_dev/superset_config.py index 7755cdb2cdc..bb76329eb91 100644 --- a/docker/pythonpath_dev/superset_config.py +++ b/docker/pythonpath_dev/superset_config.py @@ -84,6 +84,7 @@ class CeleryConfig: broker_url = f"redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_CELERY_DB}" imports = ( "superset.sql_lab", + "superset.tasks.deletion_retention", "superset.tasks.scheduler", "superset.tasks.thumbnails", "superset.tasks.cache", @@ -101,6 +102,13 @@ class CeleryConfig: "task": "reports.prune_log", "schedule": crontab(minute=10, hour=0), }, + # Gated on the SOFT_DELETE feature flag, which is off by default: the + # task is scheduled either way, but purges nothing while the flag is + # unset. Enable it in FEATURE_FLAGS below to exercise retention locally. + "deletion_retention.purge_soft_deleted": { + "task": "deletion_retention.purge_soft_deleted", + "schedule": crontab(minute=0, hour=0), + }, } diff --git a/superset/cli/deletion_retention.py b/superset/cli/deletion_retention.py new file mode 100644 index 00000000000..79453dbb2b3 --- /dev/null +++ b/superset/cli/deletion_retention.py @@ -0,0 +1,164 @@ +# 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. +"""Operator CLI for deletion retention. + +``force-purge`` and ``set-window`` are **operator-gated** — they are +protected by deployment/shell access (the ``SECURITY.md`` operator trust +boundary), not Flask-AppBuilder RBAC: a CLI invocation has no ``g.user``, so +there is no ``403`` to enforce. A future REST route would carry real +Admin RBAC. +""" + +import logging +from uuid import UUID + +import click +from flask.cli import with_appcontext + +logger = logging.getLogger(__name__) + +#: Operator-facing entity names mapped to their table. Kept as table names +#: rather than model classes so building the ``--type`` choices costs no model +#: imports at CLI start-up; the class is resolved when the option is used. +_PURGE_TYPES: dict[str, str] = { + "chart": "slices", + "dashboard": "dashboards", + "dataset": "tables", +} + + +def _resolve_model(entity_type: str | None) -> type | None: + """Map a ``--type`` value to its soft-delete model, or ``None`` for all. + + ``None`` preserves the default search across every registered model, which + is what an operator holding only a UUID has to start from. + """ + if entity_type is None: + return None + from superset.models.helpers import SoftDeleteMixin + + table = _PURGE_TYPES[entity_type.lower()] + for model in SoftDeleteMixin._registered_subclasses: # noqa: SLF001 + if getattr(model, "__tablename__", None) == table: + return model + # Unreachable while _PURGE_TYPES tracks the registered models; a mismatch + # means a model was renamed or dropped without updating the map. + raise click.ClickException( + f"No soft-delete model is registered for type {entity_type!r}." + ) + + +@click.group() +def deletion_retention() -> None: + """Manage purge of soft-deleted entities (operator-gated).""" + + +@deletion_retention.command() +@with_appcontext +@click.option( + "--days", + "-d", + required=True, + type=int, + help="Retention window in days; 0 disables.", +) +def set_window(days: int) -> None: + """Set the per-deployment retention window (SharedKey, upsert).""" + from superset.key_value.shared_entries import upsert_shared_value + from superset.key_value.types import SharedKey + + if days < 0: + raise click.BadParameter("--days must be >= 0") + upsert_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS, days) + click.echo( + f"Soft-delete retention window set to {days} day(s) for this deployment." + ) + + +@deletion_retention.command() +@with_appcontext +def show_window() -> None: + """Print the effective retention window (shared value or env fallback).""" + from superset.commands.deletion_retention.window import resolve_retention_window + + days = resolve_retention_window() + state = "disabled" if days == 0 else f"{days} day(s)" + click.echo(f"Effective soft-delete retention window: {state}.") + + +@deletion_retention.command() +@with_appcontext +@click.option( + "--uuid", + "-u", + "uuid", + required=True, + # Validate up front: a malformed value must fail with a clean + # BadParameter message, not a StatementError traceback after the + # operator has already confirmed an irreversible prompt. + type=click.UUID, + help="UUID of the entity to purge.", +) +@click.option( + "--type", + "-t", + "entity_type", + type=click.Choice(sorted(_PURGE_TYPES), case_sensitive=False), + default=None, + help=( + "Restrict the purge to one entity type. UUIDs are unique per table " + "but not across them, so a bare UUID can match more than one entity; " + "the purge refuses to guess and asks for this option." + ), +) +@click.confirmation_option( + prompt="Force-purge is irreversible — the entity and its version history " + "will be permanently removed. Continue?" +) +def force_purge(uuid: UUID, entity_type: str | None) -> None: + """Immediately and irreversibly purge an entity by UUID (compliance).""" + from superset.commands.deletion_retention.force_purge import ( + AmbiguousPurgeTargetError, + ForcePurgeCommand, + ) + + try: + result = ForcePurgeCommand( + str(uuid), model_cls=_resolve_model(entity_type) + ).run() + except AmbiguousPurgeTargetError as ex: + # The command refuses to guess between tables. Report that as a clean + # operator error naming the way out, not as a traceback -- this lands + # after the irreversible confirmation prompt has already been answered. + raise click.ClickException( + f"{ex} Re-run with --type, e.g. --type {sorted(_PURGE_TYPES)[0]}." + ) from ex + if not result.get("purged"): + if result.get("reason") == "blocked": + click.echo( + f"Entity uuid={uuid} was not purged because existing deletion " + f"rules block it: {result.get('blocked_reason')}." + ) + else: + click.echo(f"No entity found for uuid={uuid} (nothing to purge).") + return + click.echo( + f"Purged {result['entity_type']} uuid={uuid}. " + f"Dangling charts: {len(result.get('dangling_chart_uuids') or [])}; " + f"dashboard_slices removed: {result.get('removed_dashboard_slices', 0)}; " + f"version rows removed: {result.get('version_rows_removed', 0)}." + ) diff --git a/superset/commands/deletion_retention/__init__.py b/superset/commands/deletion_retention/__init__.py new file mode 100644 index 00000000000..ebff0468375 --- /dev/null +++ b/superset/commands/deletion_retention/__init__.py @@ -0,0 +1,22 @@ +# 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. +"""Deletion retention: purge of soft-deleted entities. + +Shared building blocks for the scheduled purge task +(``superset.tasks.deletion_retention``) and the operator force-purge +command, so the cascade cannot drift between the two surfaces. +""" diff --git a/superset/commands/deletion_retention/audit.py b/superset/commands/deletion_retention/audit.py new file mode 100644 index 00000000000..3c6c5bf541e --- /dev/null +++ b/superset/commands/deletion_retention/audit.py @@ -0,0 +1,238 @@ +# 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, timezone +from typing import Any, cast +from uuid import UUID + +import sqlalchemy as sa +from sqlalchemy.orm import Session, sessionmaker + +from superset import db +from superset.models.purge_audit_log import ( + PurgeAuditLog, + STATUS_BLOCKED, + STATUS_CONFIRMED, + STATUS_FAILED, + STATUS_PENDING, +) + +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)() + + +_PENDING_STALE_AFTER = timedelta(hours=1) + +TRIGGER_RETENTION = "retention" +TRIGGER_FORCE = "force" + +ACTOR_SYSTEM = "system" + + +def _utc_now() -> datetime: + """Naive UTC now for the audit columns. + + Note this deliberately differs from the metadata schema's audit + columns (``changed_on`` / ``deleted_at``), which are naive-local per + the PR #33693 UTC revert — the purge audit table is self-contained + (``created_on`` and the reconcile cutoff both use this clock), so it + can use the saner convention without a comparison hazard. + """ + return datetime.now(timezone.utc).replace(tzinfo=None) + + +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=_utc_now(), + ) + 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: + values: dict[str, Any] = {"status": status} + if status == STATUS_CONFIRMED: + values["confirmed_on"] = _utc_now() + referrers = details.get("affected_referrers") + if referrers: + values["affected_referrers"] = ",".join(referrers) + removed_dashboard_slices = details.get("removed_dashboard_slices") + if removed_dashboard_slices is not None: + values["removed_dashboard_slices"] = removed_dashboard_slices + # Conditional UPDATE, not read-then-write: only pending rows may + # transition (a delayed worker must not overwrite an outcome + # reconcile_pending() already recorded — the audit history is + # immutable once finalized), and the status predicate makes that + # atomic under concurrent finalizers. + session.execute( + sa.update(PurgeAuditLog.__table__) + .where( + PurgeAuditLog.__table__.c.id == record_id, + PurgeAuditLog.__table__.c.status == STATUS_PENDING, + ) + .values(**values) + ) + 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 _utc_now() - _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) + # Row locks so a delayed worker's finalize() (a conditional + # UPDATE on status='pending') serializes against this scan + # on PostgreSQL/MySQL; a no-op on SQLite. + .with_for_update() + .all() + ) + for record in records: + if _entity_exists(session, record) is False: + record.status = STATUS_CONFIRMED + record.confirmed_on = _utc_now() + confirmed += 1 + else: + record.status = STATUS_FAILED + failed += 1 + reconciled += 1 + session.commit() + except Exception: # pylint: disable=broad-except + session.rollback() + logger.warning( + "deletion_retention: failed to reconcile pending audit rows", + exc_info=True, + ) + finally: + session.close() + return {"reconciled": reconciled, "confirmed": confirmed, "failed": failed} diff --git a/superset/commands/deletion_retention/force_purge.py b/superset/commands/deletion_retention/force_purge.py new file mode 100644 index 00000000000..058737276bb --- /dev/null +++ b/superset/commands/deletion_retention/force_purge.py @@ -0,0 +1,199 @@ +# 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. +"""Compliance force-purge of a single entity by UUID. + +Immediate, irreversible removal of one entity regardless of the retention +window or whether it is currently soft-deleted or live. Runs the same cascade +as the time-based task with ``enforce_window=False`` — identical dependent +handling with legacy hard-delete semantics: M:N join rows hard-deleted, +a referencing live chart's loose ``datasource_id`` left dangling (the chart is +never modified). Idempotent: a UUID that resolves to nothing is a no-op. +""" + +from __future__ import annotations + +import logging +from typing import Any, cast + +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, + suppress_purge_association_versions, +) +from superset.models.helpers import skip_visibility_filter, SoftDeleteMixin + +logger: logging.Logger = logging.getLogger(__name__) + + +class AmbiguousPurgeTargetError(Exception): + """The UUID matches rows in more than one soft-delete model.""" + + +class ForcePurgeCommand: + """Force-purge the entity identified by *uuid*, if any. + + ``model_cls`` restricts resolution to a single entity type. Operators + reach this command with a bare UUID and no type, so it defaults to + searching every soft-delete model — but UUID uniqueness is only enforced + per table, so that search can be ambiguous. When it is, the command + refuses rather than guessing: picking the first match would let a + compliance deletion destroy an entity of a type nobody asked about. + + Any caller that already knows the type — notably one acting for an end + user, whose authorization was necessarily checked against one specific + entity — must pass ``model_cls`` so resolution cannot wander. + + ``require_archived`` restricts resolution to soft-deleted rows. The + cascade runs with ``enforce_window=False`` here (a force purge ignores + the retention window), which also skips its ``deleted_at`` check, so + without this a row restored between authorization and purge could be + hard-deleted while live. + """ + + def __init__( + self, + uuid: str, + actor: str = "operator", + model_cls: type[SoftDeleteMixin] | None = None, + require_archived: bool = False, + ) -> None: + self._uuid: str = uuid + self._actor: str = actor + self._model_cls = model_cls + self._require_archived = require_archived + + def _resolve(self) -> SoftDeleteMixin | None: + """Find the entity by UUID, visibility-filter bypassed. + + Searches only ``model_cls`` when given, else every registered + soft-delete model — raising :class:`AmbiguousPurgeTargetError` if more + than one model matches. Matches live rows as well as soft-deleted ones + unless ``require_archived`` is set. + """ + candidates = ( + [self._model_cls] + if self._model_cls is not None + else SoftDeleteMixin._registered_subclasses # noqa: SLF001 + ) + matches: list[SoftDeleteMixin] = [] + for model in candidates: + if not hasattr(model, "uuid"): + continue + with skip_visibility_filter(db.session, model): + query = db.session.query(model).filter(model.uuid == self._uuid) + if self._require_archived: + query = query.filter(model.deleted_at.is_not(None)) + entity = query.first() + if entity is not None: + matches.append(entity) + if len(matches) > 1: + raise AmbiguousPurgeTargetError( + f"uuid={self._uuid!r} matches " + f"{', '.join(sorted(type(m).__name__ for m in matches))}; " + "pass the entity type to disambiguate" + ) + return matches[0] if matches else None + + def run(self) -> dict[str, Any]: + """Resolve + purge. Returns a summary; a no-op when nothing matches.""" + audit.reconcile_pending() + entity = self._resolve() + if entity is None: + logger.info("force_purge: no entity for uuid=%s (no-op)", self._uuid) + return {"purged": False, "reason": "not_found", "uuid": self._uuid} + + entity_type = str(cast(Any, type(entity)).__tablename__) + removed_dashboard_slices = dashboard_slice_count(db.session, entity) + # The audit row commits independently. Release the resolving read + # transaction first, then resolve again against post-audit state. + db.session.rollback() # pylint: disable=consider-using-transaction + record_id = audit.write_ahead( + trigger=audit.TRIGGER_FORCE, + actor=self._actor, + entity_type=entity_type, + entity_uuid=self._uuid, + removed_dashboard_slices=removed_dashboard_slices, + ) + entity = self._resolve() + if entity is None: + audit.fail(record_id) + logger.info( + "force_purge: entity disappeared before purge uuid=%s (no-op)", + self._uuid, + ) + return {"purged": False, "reason": "not_found", "uuid": self._uuid} + try: + with suppress_purge_association_versions(db.session): + result: CascadeResult = cascade_hard_delete( + db.session, entity, enforce_window=False + ) + # Commit AFTER the suppression block: Continuum executes its + # pending association statements during flush/commit, so the + # block's exit-time trim must run first or a session carrying + # versioned state would write the purge-queued shadows anyway. + # Commit/rollback are managed manually so audit.fail() can + # record the outcome after the purge transaction resolves. + db.session.commit() # pylint: disable=consider-using-transaction + except Exception: + db.session.rollback() # pylint: disable=consider-using-transaction + audit.fail(record_id) + raise + if result.purged: + audit.confirm( + record_id, + affected_referrers=result.dangling_chart_uuids, + removed_dashboard_slices=result.removed_dashboard_slices, + ) + elif result.blocked_reason is not None: + audit.block(record_id) + else: + audit.fail(record_id) + if result.purged: + logger.info( + "force_purge: purged %s uuid=%s " + "(dangling charts=%d, dashboard_slices=%d)", + result.entity_type, + self._uuid, + len(result.dangling_chart_uuids), + result.removed_dashboard_slices, + ) + elif result.blocked_reason is not None: + logger.info( + "force_purge: blocked %s uuid=%s reason=%s", + result.entity_type, + self._uuid, + result.blocked_reason, + ) + else: + logger.info( + "force_purge: no-op %s uuid=%s (restored or already gone)", + result.entity_type, + self._uuid, + ) + return { + "purged": result.purged, + "reason": "blocked" if result.blocked_reason is not None else None, + "blocked_reason": result.blocked_reason, + "entity_type": result.entity_type, + "uuid": self._uuid, + "dangling_chart_uuids": result.dangling_chart_uuids, + "removed_dashboard_slices": result.removed_dashboard_slices, + "version_rows_removed": result.version_rows_removed, + } diff --git a/superset/commands/deletion_retention/purge_cascade.py b/superset/commands/deletion_retention/purge_cascade.py new file mode 100644 index 00000000000..3e8b2b45196 --- /dev/null +++ b/superset/commands/deletion_retention/purge_cascade.py @@ -0,0 +1,607 @@ +# 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 _identity_predicates(table: sa.Table, entity_id: int, entity: Any) -> list[Any]: + """Predicates pinning a statement to the exact row *entity* came from. + + The id alone is not an identity: a row can be removed and its id handed to + a different entity, and both SQLite rowids and a sequence wrapped by an + operator will reuse values. Callers snapshot the uuid well before the + cascade runs -- the retention task writes an audit row in between -- so the + locked claim and the conditional delete carry the uuid too, and a reused id + simply matches nothing rather than purging a stranger under the snapshot's + name. Checking it before the lock would only narrow that window; a + predicate on the claim closes it. + + Falls back to the id alone for a model with no uuid column. + """ + predicates: list[Any] = [table.c.id == entity_id] + uuid_column = table.c.get("uuid") + expected_uuid = entity_uuid(entity) + if uuid_column is not None and expected_uuid is not None: + predicates.append(uuid_column == expected_uuid) + return predicates + + +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(): + identity = _identity_predicates(table, entity_id, entity) + claim = sa.select(table.c.id).where(*identity) + 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) + + delete_entity = sa.delete(table).where(*identity) + if enforce_window: + delete_entity = delete_entity.where( + table.c.deleted_at.is_not(None) + ).where(table.c.deleted_at < cutoff) + if session.execute(delete_entity).rowcount == 0: + raise PurgeRaceLostError + + if permission_name is not None: + _cleanup_dataset_permission(session, permission_name, entity_id) + except PurgeRaceLostError: + logger.info( + "deletion_retention: %s id=%s not purged (restored or already gone)", + entity_type, + entity_id, + ) + return CascadeResult(purged=False, entity_type=entity_type, entity_uuid=uuid) + except (PurgeBlockedError, IntegrityError) as ex: + logger.info( + "deletion_retention: %s id=%s blocked by existing deletion rules", + entity_type, + entity_id, + ) + return CascadeResult( + purged=False, + entity_type=entity_type, + entity_uuid=uuid, + blocked_reason=str(ex), + ) + + return CascadeResult( + purged=True, + entity_type=entity_type, + entity_uuid=uuid, + dangling_chart_uuids=dangling_chart_uuids, + removed_dashboard_slices=removed_dashboard_slices, + version_rows_removed=version_rows, + ) + + +_USER_FACING_TYPE: dict[str, str] = { + "slices": "chart", + "dashboards": "dashboard", + "tables": "dataset", +} + + +def _validate_deletion_allowed( + session: Session, model: type[Any], entity_id: int +) -> None: + """Apply the dependency guards used by ordinary delete commands.""" + # pylint: disable=import-outside-toplevel + from superset.models.dashboard import Dashboard + from superset.models.slice import Slice + from superset.reports.models import ReportSchedule + + column: Any | None = None + if model is Slice: + column = ReportSchedule.chart_id + elif model is Dashboard: + column = ReportSchedule.dashboard_id + if ( + column is not None + and session.execute( + sa.select(ReportSchedule.id).where(column == entity_id).limit(1) + ).first() + ): + raise PurgeBlockedError("associated alerts or reports exist") + + +def _count_dashboard_slices(session: Session, model: type[Any], entity_id: int) -> int: + """Snapshot relationship counts before DB cascades can remove rows.""" + # pylint: disable=import-outside-toplevel + from superset.models.dashboard import Dashboard, dashboard_slices + from superset.models.slice import Slice + + predicate: Any | None = None + if model is Dashboard: + predicate = dashboard_slices.c.dashboard_id == entity_id + elif model is Slice: + predicate = dashboard_slices.c.slice_id == entity_id + if predicate is None: + return 0 + return int( + session.execute( + sa.select(sa.func.count()).select_from(dashboard_slices).where(predicate) + ).scalar_one() + ) + + +def dashboard_slice_count(session: Session, entity: Any) -> int: + """Return the current dashboard relationship count for audit write-ahead.""" + return _count_dashboard_slices(session, type(entity), entity.id) + + +def _delete_m2m_joins(session: Session, model: type[Any], entity_id: int) -> None: + """Hard-delete every M:N join / association row the entity owns. + + Relationship counts are captured before this function runs so database + cascades cannot make the reported values dialect-dependent. + """ + # pylint: disable=import-outside-toplevel + from superset.connectors.sqla.models import SqlaTable + from superset.models.dashboard import Dashboard, dashboard_slices + from superset.models.slice import Slice + from superset.subjects.models import ( + chart_editors, + chart_viewers, + dashboard_editors, + dashboard_viewers, + sqlatable_editors, + ) + from superset.tags.models import ObjectType, TaggedObject + + if model is Dashboard: + session.execute( + sa.delete(dashboard_slices).where( + dashboard_slices.c.dashboard_id == entity_id + ) + ) + for association in (dashboard_editors, dashboard_viewers): + session.execute( + sa.delete(association).where(association.c.dashboard_id == entity_id) + ) + _delete_tags(session, TaggedObject, ObjectType.dashboard, entity_id) + elif model is Slice: + # Every dashboard_slices row pointing at this chart, including those + # owned by live dashboards (the live dashboard survives, minus this + # chart from its layout). + session.execute( + sa.delete(dashboard_slices).where(dashboard_slices.c.slice_id == entity_id) + ) + for association in (chart_editors, chart_viewers): + session.execute( + sa.delete(association).where(association.c.chart_id == entity_id) + ) + _delete_tags(session, TaggedObject, ObjectType.chart, entity_id) + elif model is SqlaTable: + from superset.connectors.sqla.models import RLSFilterTables + + session.execute( + sa.delete(sqlatable_editors).where( + sqlatable_editors.c.table_id == entity_id + ) + ) + session.execute( + sa.delete(RLSFilterTables).where(RLSFilterTables.c.table_id == entity_id) + ) + _delete_tags(session, TaggedObject, ObjectType.dataset, entity_id) + + +def _delete_tags( + session: Session, tagged_object: type[Any], object_type: Any, entity_id: int +) -> None: + """Remove ``tagged_object`` rows skipped by the Core bulk delete.""" + session.execute( + sa.delete(tagged_object.__table__).where( + tagged_object.object_id == entity_id, + tagged_object.object_type == object_type, + ) + ) + + +def _delete_owned_children(session: Session, model: type[Any], entity_id: int) -> None: + """Hard-delete the entity's owned children — rows with no independent + existence: a dataset's columns and metrics, a dashboard's embedded + configs. Charts have no such owned child tables today. + """ + # pylint: disable=import-outside-toplevel + from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn + from superset.models.dashboard import Dashboard + from superset.models.embedded_dashboard import EmbeddedDashboard + + if model is SqlaTable: + session.execute( + sa.delete(TableColumn.__table__).where( + TableColumn.__table__.c.table_id == entity_id + ) + ) + session.execute( + sa.delete(SqlMetric.__table__).where( + SqlMetric.__table__.c.table_id == entity_id + ) + ) + elif model is Dashboard: + # Embedded configs (delete-orphan children carrying the public + # embed UUID and allowed_domains) — the ORM cascade does not fire + # for Core deletes and the DB cascade is a backstop only. + session.execute( + sa.delete(EmbeddedDashboard.__table__).where( + EmbeddedDashboard.__table__.c.dashboard_id == entity_id + ) + ) + + +def _dataset_permission_name(entity: Any) -> str: + """Capture the permission identifier while dataset attributes are readable.""" + # pylint: disable=import-outside-toplevel + from superset import security_manager + + return str( + security_manager.get_dataset_perm( + entity.id, entity.table_name, entity.database.database_name + ) + ) + + +def _cleanup_dataset_permission( + session: Session, permission_name: str, entity_id: int +) -> None: + """Replicate ``SqlaTable.after_delete`` permission cleanup. + + Core ``sa.delete`` does not fire the ORM ``after_delete`` listener that + normally removes the dataset's ``datasource access`` view-menu / + permission-view, so it is done explicitly here or the PVM is orphaned. + """ + # pylint: disable=import-outside-toplevel + from superset import security_manager + + security_manager._delete_pvm_on_sqla_event( # pylint: disable=protected-access + None, session.connection(), "datasource_access", permission_name + ) + logger.debug("deletion_retention: removed dataset permission for id=%s", entity_id) + + +def _entity_version_targets( + model: type[Any], + metadata: sa.MetaData, + parent_shadow: sa.Table, + entity_id: int, +) -> list[tuple[sa.Table, Any]]: + """The ``(shadow_table, row_predicate)`` pairs that make up *this* entity's + own version history: the parent shadow keyed by ``id``, plus — per type — + the dashboard/chart M2M shadow (``dashboard_slices_version``) and a + dataset's child shadows (``table_columns_version`` / ``sql_metrics_version`` + keyed by ``table_id``). It never touches another entity's rows.""" + # pylint: disable=import-outside-toplevel + from superset.connectors.sqla.models import SqlaTable + from superset.models.dashboard import Dashboard + from superset.models.slice import Slice + + targets: list[tuple[sa.Table, Any]] = [ + (parent_shadow, parent_shadow.c.id == entity_id) + ] + m2m = metadata.tables.get("dashboard_slices_version") + if m2m is not None and model is Dashboard: + targets.append((m2m, m2m.c.dashboard_id == entity_id)) + elif m2m is not None and model is Slice: + targets.append((m2m, m2m.c.slice_id == entity_id)) + elif model is SqlaTable: + for child_name in ("table_columns_version", "sql_metrics_version"): + child = metadata.tables.get(child_name) + if child is not None and "table_id" in child.c: + targets.append((child, child.c.table_id == entity_id)) + return targets + + +def _delete_version_history(session: Session, entity: Any, entity_id: int) -> int: + """Remove the entity's own version-history rows. + + Deletes the entity's parent + child + M2M ``*_version`` shadow rows and the + ``version_changes`` rows scoped to this entity (by ``entity_kind`` + + ``entity_id`` — **not** by transaction, which is shared), then sweeps any + ``version_transaction`` left owning zero surviving shadow / change rows. A + ``version_transaction`` is a *shared* unit-of-work boundary that can span + entities, so it is removed only once orphaned — never blind-deleted. Gated + on the version tables existing, so it no-ops cleanly without versioning. + """ + if not _version_tables_present(session.get_bind()): + return 0 + + # pylint: disable=import-outside-toplevel + try: + from sqlalchemy_continuum import version_class, versioning_manager + from sqlalchemy_continuum.exc import ClassNotVersioned + except ImportError: + return 0 + + model = type(entity) + try: + parent_shadow = version_class(model).__table__ + except ClassNotVersioned: + return 0 + + metadata = parent_shadow.metadata + # version_transaction lives in Continuum's manager, not the shadow metadata. + tx = versioning_manager.transaction_cls.__table__ + changes = metadata.tables.get("version_changes") + targets = _entity_version_targets(model, metadata, parent_shadow, entity_id) + + # Transactions these shadow rows are anchored to — orphan-sweep candidates. + tx_ids: set[int] = set() + for tbl, pred in targets: + if "transaction_id" in tbl.c: + tx_ids.update( + row[0] + for row in session.execute( + sa.select(tbl.c.transaction_id).where(pred).distinct() + ) + ) + + removed = 0 + for tbl, pred in targets: + removed += session.execute(sa.delete(tbl).where(pred)).rowcount + + # version_changes scoped to this entity (entity_kind + entity_id). + # Counted alongside the shadow rows: callers report the total as the + # definitive number of version-history rows removed. + if changes is not None: + from superset.versioning.changes import ENTITY_KIND_BY_CLASS_NAME + + kind = ENTITY_KIND_BY_CLASS_NAME.get(model.__name__) + if kind is not None: + removed += session.execute( + sa.delete(changes).where( + changes.c.entity_kind == kind, + changes.c.entity_id == entity_id, + ) + ).rowcount + + if tx is not None and tx_ids: + _sweep_orphan_transactions(session, metadata, tx, changes, tx_ids) + + return removed + + +# Bound for IN() membership lists, below SQLite's historical 999 +# bind-variable limit (well under PostgreSQL and MySQL limits). An entity +# with a long edit history can anchor thousands of transactions; unchunked +# IN lists would fail the sweep on SQLite — and a failing sweep makes the +# entity permanently unpurgeable. +_IN_CLAUSE_CHUNK: int = 500 + + +def _chunked(values: set[int]) -> Iterator[list[int]]: + """Yield sorted ``_IN_CLAUSE_CHUNK``-sized slices of *values*.""" + ordered = sorted(values) + for start in range(0, len(ordered), _IN_CLAUSE_CHUNK): + yield ordered[start : start + _IN_CLAUSE_CHUNK] + + +def _sweep_orphan_transactions( + session: Session, + metadata: sa.MetaData, + tx: sa.Table, + changes: sa.Table | None, + tx_ids: set[int], +) -> None: + """Delete unreferenced ``version_transaction`` rows. + + Every ``*_version`` table and ``version_changes`` is checked. A double-sweep + of the same orphan is a harmless no-op. + + A shadow row references a transaction through **either** endpoint of its + lifespan: ``transaction_id`` (created at) or ``end_transaction_id`` (closed + at). Both are foreign keys, so both keep a transaction alive. + + Counting only the created-at side is a live FK hazard whenever a + transaction spans entities, which is the normal case: one flush saving a + dashboard, chart and dataset gives all three a shadow row at ``tx=X``. If + the dashboard alone is later edited at ``tx=Y``, its row at ``tx=X`` closes + with ``end_transaction_id=Y`` while the others stay live at ``tx=X``. + Purging the dashboard removes its own shadow rows, after which nothing + *creates* at ``tx=Y`` — but the sibling rows still point at it, or at ``X``, + through their end column. Deleting the transaction then fails the foreign + key, and the caller reports the entity as blocked by deletion rules rather + than as the incomplete cascade it is. ``version_history_retention`` hit the + same trap and documents it on ``_delete_for_transactions``. + """ + sources = [ + t + for name, t in metadata.tables.items() + if name.endswith("_version") and "transaction_id" in t.c + ] + if changes is not None and "transaction_id" in changes.c: + sources.append(changes) + still_referenced: set[int] = set() + for source in sources: + # ``version_changes`` has no closing endpoint; shadow tables do. + columns = [ + source.c[name] + for name in ("transaction_id", "end_transaction_id") + if name in source.c + ] + for column in columns: + for chunk in _chunked(tx_ids): + still_referenced.update( + row[0] + for row in session.execute( + sa.select(column).where(column.in_(chunk)).distinct() + ) + ) + if orphaned := tx_ids - still_referenced: + for chunk in _chunked(orphaned): + session.execute(sa.delete(tx).where(tx.c.id.in_(chunk))) diff --git a/superset/commands/deletion_retention/window.py b/superset/commands/deletion_retention/window.py new file mode 100644 index 00000000000..e64bc90d8eb --- /dev/null +++ b/superset/commands/deletion_retention/window.py @@ -0,0 +1,81 @@ +# 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. +"""Resolve the soft-delete retention window.""" + +from __future__ import annotations + +import logging + +from flask import current_app + +from superset.key_value.shared_entries import get_shared_value +from superset.key_value.types import SharedKey + +logger: logging.Logger = logging.getLogger(__name__) + +_DEFAULT_RETENTION_DAYS: int = 30 + + +def _config_retention_days() -> int: + """Return a validated config fallback without breaking scheduled runs.""" + configured = current_app.config.get( + "SOFT_DELETE_RETENTION_DAYS", _DEFAULT_RETENTION_DAYS + ) + try: + if isinstance(configured, bool): + raise ValueError + days = int(configured) + if days < 0: + raise ValueError + return days + except (TypeError, ValueError): + logger.warning( + "deletion_retention: ignoring malformed config retention value %r; " + "falling back to %d days", + configured, + _DEFAULT_RETENTION_DAYS, + ) + return _DEFAULT_RETENTION_DAYS + + +def resolve_retention_window() -> int: + """Return the retention window in days, read live on each call. + + Resolution order: + + 1. The per-deployment value persisted under + ``SharedKey.SOFT_DELETE_RETENTION_DAYS`` (read live; takes + precedence when present). + 2. Otherwise the ``SOFT_DELETE_RETENTION_DAYS`` config / + environment seed default (itself defaulting to 30). + + ``0`` from either source is a meaningful "disable", so the shared + value is selected with an explicit ``is None`` check — never ``or``, + which would treat ``0`` as unset. A malformed shared value is + rejected (logged) and the fallback is used rather than crashing the + scheduled task. + """ + if (shared := get_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS)) is not None: + if isinstance(shared, bool) or not isinstance(shared, int) or shared < 0: + logger.warning( + "deletion_retention: ignoring malformed shared retention value %r; " + "falling back to config", + shared, + ) + else: + return shared + return _config_retention_days() diff --git a/superset/config.py b/superset/config.py index 70c932f611f..82f14935754 100644 --- a/superset/config.py +++ b/superset/config.py @@ -997,6 +997,12 @@ USER_AGENT_FUNC: Callable[[Database, utils.QuerySource | None], str] | None = No # This is merely a default. FEATURE_FLAGS: dict[str, bool] = {} +# Retention policy for soft-deleted dashboards, charts, and datasets. A value of +# zero disables scheduled purging. Dry-run mode is enabled by default so operators +# must explicitly opt in to irreversible deletion. +SOFT_DELETE_RETENTION_DAYS: int = 30 +SOFT_DELETE_PURGE_DRY_RUN: bool = True + # A function that receives a dict of all feature flags # (DEFAULT_FEATURE_FLAGS merged with FEATURE_FLAGS) # can alter it, and returns a similar dict. Note the dict of feature @@ -1758,6 +1764,7 @@ class CeleryConfig: # pylint: disable=too-few-public-methods broker_url = "sqla+sqlite:///celerydb.sqlite" imports = ( "superset.sql_lab", + "superset.tasks.deletion_retention", "superset.tasks.scheduler", "superset.tasks.thumbnails", "superset.tasks.cache", @@ -1790,6 +1797,10 @@ class CeleryConfig: # pylint: disable=too-few-public-methods "task": "version_history.prune_old_versions", "schedule": crontab(minute=0, hour=3), }, + "deletion_retention.purge_soft_deleted": { + "task": "deletion_retention.purge_soft_deleted", + "schedule": crontab(minute=0, hour=0), + }, # Uncomment to enable pruning of the query table # "prune_query": { # "task": "prune_query", diff --git a/superset/key_value/types.py b/superset/key_value/types.py index fbc6cc56526..b1f27de4cbf 100644 --- a/superset/key_value/types.py +++ b/superset/key_value/types.py @@ -56,6 +56,10 @@ class SharedKey(StrEnum): # Monotonically increasing version used to revoke outstanding guest tokens. # Bumping it invalidates every guest token minted with a lower version. GUEST_TOKEN_REVOCATION_VERSION = "guest_token_revocation_version" # noqa: S105 + # Per-deployment retention window (days) for purging soft-deleted entities. + # Read live each run by deletion_retention.purge_soft_deleted; falls back to + # SOFT_DELETE_RETENTION_DAYS when unset. 0 disables the purge. + SOFT_DELETE_RETENTION_DAYS = "soft_delete_retention_days" class KeyValueCodec(ABC): diff --git a/superset/migrations/versions/2026-07-28_09-00_e7d93a524ff6_add_purge_audit_log.py b/superset/migrations/versions/2026-07-28_09-00_e7d93a524ff6_add_purge_audit_log.py new file mode 100644 index 00000000000..9732955cbbc --- /dev/null +++ b/superset/migrations/versions/2026-07-28_09-00_e7d93a524ff6_add_purge_audit_log.py @@ -0,0 +1,88 @@ +# 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. +"""add purge_audit_log + +Immutable, content-free audit record for deletion-retention purges. Survives +the entity it names; written write-ahead (pending -> confirmed). + +Revision ID: e7d93a524ff6 +Revises: d3b9a1f6c204 +Create Date: 2026-07-28 09:00:00.000000 + +""" + +import sqlalchemy as sa +from sqlalchemy_utils import UUIDType + +from superset.migrations.shared.utils import ( + create_index, + create_table, + drop_index, + drop_table, +) + +# revision identifiers, used by Alembic. +revision = "e7d93a524ff6" +down_revision = "d3b9a1f6c204" + + +def upgrade() -> None: + create_table( + "purge_audit_log", + sa.Column("id", UUIDType(binary=True), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("trigger", sa.String(length=16), nullable=False), + sa.Column("actor", sa.String(length=256), nullable=False), + sa.Column("entity_type", sa.String(length=64), nullable=False), + sa.Column("entity_uuid", sa.String(length=36), nullable=True), + sa.Column("affected_referrers", sa.Text(), nullable=True), + sa.Column( + "removed_dashboard_slices", + sa.Integer(), + server_default="0", + nullable=False, + ), + sa.Column("created_on", sa.DateTime(), nullable=False), + sa.Column("confirmed_on", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + create_index( + "purge_audit_log", + "ix_purge_audit_log_entity_uuid", + ["entity_uuid"], + ) + create_index( + "purge_audit_log", + "ix_purge_audit_log_status_created_on", + ["status", "created_on"], + ) + # Chart purges delete dashboard_slices_version rows by slice_id; the + # table's PK and existing indexes all lead with other columns, so + # without this every purged chart full-scans the association shadow + # history. + create_index( + "dashboard_slices_version", + "ix_dashboard_slices_version_slice_id", + ["slice_id"], + ) + + +def downgrade() -> None: + drop_index("dashboard_slices_version", "ix_dashboard_slices_version_slice_id") + drop_index("purge_audit_log", "ix_purge_audit_log_status_created_on") + drop_index("purge_audit_log", "ix_purge_audit_log_entity_uuid") + drop_table("purge_audit_log") diff --git a/superset/models/__init__.py b/superset/models/__init__.py index 750c61ccc23..1db3711227c 100644 --- a/superset/models/__init__.py +++ b/superset/models/__init__.py @@ -14,4 +14,10 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -from . import core, dynamic_plugins, sql_lab, user_attributes # noqa: F401 +from . import ( # noqa: F401 + core, + dynamic_plugins, + purge_audit_log, + sql_lab, + user_attributes, +) diff --git a/superset/models/purge_audit_log.py b/superset/models/purge_audit_log.py new file mode 100644 index 00000000000..438500e90fd --- /dev/null +++ b/superset/models/purge_audit_log.py @@ -0,0 +1,60 @@ +# 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. +"""The ``purge_audit_log`` table: immutable, content-free purge records. + +The write-ahead audit protocol that populates this table lives in +:mod:`superset.commands.deletion_retention.audit`; the model is defined +here so it registers with ``Model.metadata`` at app init like every other +Superset model (Alembic autogenerate and metadata-driven tooling would +otherwise not see the table). +""" + +from uuid import uuid4 + +import sqlalchemy as sa +from flask_appbuilder import Model +from sqlalchemy import Column, DateTime, Integer, String, Text +from sqlalchemy_utils import UUIDType + +STATUS_PENDING = "pending" +STATUS_CONFIRMED = "confirmed" +STATUS_FAILED = "failed" +STATUS_BLOCKED = "blocked" + + +class PurgeAuditLog(Model): + """Immutable, content-free record of a purge.""" + + __tablename__ = "purge_audit_log" + __table_args__ = ( + # Backs reconcile_pending()'s stale-pending scan; mirrors the + # index created by migration e7d93a524ff6. + sa.Index("ix_purge_audit_log_status_created_on", "status", "created_on"), + ) + + 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) diff --git a/superset/tasks/deletion_retention.py b/superset/tasks/deletion_retention.py new file mode 100644 index 00000000000..efb7119877d --- /dev/null +++ b/superset/tasks/deletion_retention.py @@ -0,0 +1,307 @@ +# 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" +# Batch window for the eligible-id scan (SELECT ... LIMIT): bounds how many +# entities one iteration holds eligible before purging them one at a time. +_BATCH: int = 500 + + +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 _reconcile_unless_dry_run(dry_run: bool) -> None: + """Finalize stale pending audit rows, except during a dry run. + + Reconciliation is a durable write. A dry run is documented as reporting + what *would* happen, so it must not resolve another run's audit attempts + as a side effect — an operator sizing up a rollout would otherwise change + the very record they are inspecting. + """ + if not dry_run: + audit.reconcile_pending() + + +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} + + # Same clock as SoftDeleteMixin.soft_delete(): deleted_at is stamped + # with naive-local datetime.now() (mirroring changed_on per the + # PR #33693 UTC revert), so the cutoff must be naive-local too — a + # UTC-derived cutoff would shift the retention window by the server's + # timezone offset, purging early west of UTC. If deleted_at ever moves + # to UTC-aware, this must move with it. + cutoff = datetime.now() - timedelta(days=window_days) + _reconcile_unless_dry_run(dry_run) + purged: dict[str, int] = {} + would_purge: dict[str, int] = {} + failures = 0 + blocked = 0 + + for model in _soft_delete_models(): + entity_type = _model_table_name(model) + purged_n, would_n, failed_n, blocked_n = _purge_model(model, cutoff, dry_run) + if would_n: + would_purge[entity_type] = would_n + if purged_n: + purged[entity_type] = purged_n + failures += failed_n + blocked += blocked_n + + if dry_run: + for entity_type, count in would_purge.items(): + stats_logger_manager.instance.gauge( + f"{_METRIC_PREFIX}.would_purge.{entity_type}", count + ) + logger.info("deletion_retention: DRY RUN would_purge=%s", would_purge) + return {"dry_run": 1, "would_purge": would_purge} + + for entity_type, count in purged.items(): + stats_logger_manager.instance.gauge( + f"{_METRIC_PREFIX}.purged.{entity_type}", count + ) + if failures: + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.cascade_failures") + if blocked: + stats_logger_manager.instance.gauge( + f"{_METRIC_PREFIX}.blocked_by_reference", blocked + ) + stats = { + "purged": purged, + "cascade_failures": failures, + "blocked_by_reference": blocked, + } + logger.info("deletion_retention: %s", stats) + return stats + + +def _purge_model( + model: type[SoftDeleteMixin], cutoff: datetime, dry_run: bool +) -> tuple[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.""" + 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 + + +def _purge_one( + model: type[SoftDeleteMixin], entity_id: int, cutoff: datetime +) -> CascadeResult | None: + """Purge a single entity in its own transaction with a write-ahead audit.""" + with skip_visibility_filter(db.session, model): + entity = db.session.get(model, entity_id) + if entity is None: + return None + entity_uuid_value = entity_uuid(entity) + removed_dashboard_slices = dashboard_slice_count(db.session, entity) + # The audit row commits on a separate connection. End the read transaction + # before that write so SQLite can later promote this session to a writer. + # Re-resolving below also ensures the cascade acts on post-audit state. + db.session.rollback() # pylint: disable=consider-using-transaction + record_id = audit.write_ahead( + trigger=audit.TRIGGER_RETENTION, + actor=audit.ACTOR_SYSTEM, + entity_type=_model_table_name(model), + entity_uuid=entity_uuid_value, + removed_dashboard_slices=removed_dashboard_slices, + ) + if record_id is None: + # Fail closed: the scheduled purge must not delete unauditably. + # The entity stays soft-deleted and is retried next run; the + # operator-invoked force-purge makes the opposite call (deletion + # outranks audit when a human is present). + raise RuntimeError( + f"deletion_retention: write-ahead audit failed for " + f"{_model_table_name(model)} id={entity_id}; skipping purge" + ) + with skip_visibility_filter(db.session, model): + entity = db.session.get(model, entity_id) + if entity is None: + audit.fail(record_id) + return None + if entity_uuid(entity) != entity_uuid_value: + # The row under this id is not the row the audit describes. The + # cascade's conditional claim would still refuse to purge anything + # live, so this is an attribution guard rather than a destructive + # one: without it, an id reused between the snapshot and here (SQLite + # recycles rowids; sequences do not) could purge a genuinely eligible + # entity while the audit names a different one. An audit row that + # identifies the wrong object is worse than a skipped purge. + logger.warning( + "deletion_retention: %s id=%s changed identity before purge " + "(expected uuid=%s); skipping", + _model_table_name(model), + entity_id, + entity_uuid_value, + ) + audit.fail(record_id) + return None + try: + with suppress_purge_association_versions(db.session): + result = cascade_hard_delete( + db.session, entity, enforce_window=True, cutoff=cutoff + ) + # Commit AFTER the suppression block: Continuum executes its + # pending association statements during flush/commit, so the + # block's exit-time trim must run first or a session carrying + # versioned state would write the purge-queued shadows anyway. + # Commit/rollback are managed manually so audit.fail() can + # record the outcome after the purge transaction resolves. + db.session.commit() # pylint: disable=consider-using-transaction + except Exception: + db.session.rollback() # pylint: disable=consider-using-transaction + audit.fail(record_id) + raise + if result.purged: + audit.confirm( + record_id, + affected_referrers=result.dangling_chart_uuids, + removed_dashboard_slices=result.removed_dashboard_slices, + ) + elif result.blocked_reason is not None: + audit.block(record_id) + else: + audit.fail(record_id) + return result + + +@celery_app.task(name="deletion_retention.purge_soft_deleted") +def purge_soft_deleted() -> dict[str, Any]: + """Beat entry point. Resolves the window live, honors the SOFT_DELETE + rollout gate and dry-run flag, and isolates failures so one bad run does + not poison the schedule.""" + # While the temporary SOFT_DELETE rollout gate is off the delete path + # writes no ``deleted_at`` rows, so the task already no-ops; check the gate + # explicitly for clarity (the check is removed when the gate is). + if not feature_flag_manager.is_feature_enabled("SOFT_DELETE"): + logger.info("deletion_retention: SOFT_DELETE gate off; skipping") + return {"skipped": 1} + window_days = resolve_retention_window() + dry_run = bool(current_app.config.get("SOFT_DELETE_PURGE_DRY_RUN", True)) + try: + return _purge_impl(window_days, dry_run) + except Exception: # pylint: disable=broad-except + logger.exception("deletion_retention.purge_soft_deleted: task failed") + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.failed") + return {"error": 1} diff --git a/tests/integration_tests/deletion_retention/__init__.py b/tests/integration_tests/deletion_retention/__init__.py new file mode 100644 index 00000000000..13a83393a91 --- /dev/null +++ b/tests/integration_tests/deletion_retention/__init__.py @@ -0,0 +1,16 @@ +# 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. diff --git a/tests/integration_tests/deletion_retention/_base.py b/tests/integration_tests/deletion_retention/_base.py new file mode 100644 index 00000000000..2d6c92db621 --- /dev/null +++ b/tests/integration_tests/deletion_retention/_base.py @@ -0,0 +1,255 @@ +# 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 base + self-contained builders for deletion-retention tests. + +These tests do not depend on the example datasets — each builds its own +``Database`` + ``SqlaTable`` so they run on a bare (schema-only) test DB. +Everything created is torn down (bypassing the soft-delete visibility +filter) so a leftover soft-deleted row never trips a later test. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import Any + +import sqlalchemy as sa + +from superset import db +from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn +from superset.constants import SKIP_VISIBILITY_FILTER_CLASSES +from superset.models.core import Database +from superset.models.dashboard import Dashboard +from superset.models.slice import Slice +from tests.integration_tests.base_tests import SupersetTestCase + +_PREFIX = "retention_it_" + + +def _bypass(model: type[Any]) -> dict[str, Any]: + return {"execution_options": {SKIP_VISIBILITY_FILTER_CLASSES: {model}}} + + +class DeletionRetentionTestBase(SupersetTestCase): + """Builds an isolated database + dataset and cleans up after itself.""" + + def setUp(self) -> None: + super().setUp() + self._cleanup() + self.database: Database = Database( + database_name=f"{_PREFIX}db", sqlalchemy_uri="sqlite://" + ) + db.session.add(self.database) + db.session.commit() + self.dataset: SqlaTable = self.make_dataset("ds") + + def tearDown(self) -> None: + self._cleanup() + super().tearDown() + + # -- builders ----------------------------------------------------------- + + def make_dataset(self, name: str, with_children: bool = False) -> SqlaTable: + ds = SqlaTable(table_name=f"{_PREFIX}{name}", database=self.database) + db.session.add(ds) + db.session.commit() + if with_children: + db.session.add(TableColumn(column_name=f"{_PREFIX}col", table=ds)) + db.session.add( + SqlMetric( + metric_name=f"{_PREFIX}metric", expression="count(*)", table=ds + ) + ) + db.session.commit() + return ds + + def make_chart(self, name: str, dataset: SqlaTable | None = None) -> Slice: + dataset = dataset or self.dataset + chart = Slice( + slice_name=f"{_PREFIX}{name}", + datasource_type="table", + datasource_id=dataset.id, + viz_type="table", + ) + db.session.add(chart) + db.session.commit() + return chart + + def make_dashboard(self, name: str, slices: list[Slice] | None = None) -> Dashboard: + dash = Dashboard( + dashboard_title=f"{_PREFIX}{name}", + slug=f"{_PREFIX}{name}", + slices=slices or [], + ) + db.session.add(dash) + db.session.commit() + return dash + + def soft_delete(self, entity: Any, days_ago: int) -> None: + """Mark *entity* soft-deleted with a backdated ``deleted_at``.""" + entity.deleted_at = datetime.now() - timedelta(days=days_ago) + db.session.add(entity) + db.session.commit() + + # -- assertions / lookups ---------------------------------------------- + + def exists(self, model: type[Any], entity_id: int) -> bool: + row = ( + db.session.query(model) + .execution_options(**{SKIP_VISIBILITY_FILTER_CLASSES: {model}}) + .filter(model.id == entity_id) + .one_or_none() + ) + return row is not None + + def count(self, sql: str, params: dict[str, Any]) -> int: + return db.session.execute(sa.text(sql), params).scalar() or 0 + + # -- version-history forging ------------------------------------------- + + def forge_version_row( + self, + model: type[Any], + entity_id: int, + tx_id: int, + end_tx_id: int | None = None, + ) -> None: + """Insert a version_transaction + parent shadow + version_changes row + for *entity_id* anchored at *tx_id* (so a purge has history to remove + without needing live capture to be enabled). + + Pass *end_tx_id* to close the shadow row at another transaction, which + is how a real edit ends the previous row's lifespan. Both endpoints are + foreign keys to ``version_transaction``, so a closed row keeps the + transaction it closes at alive just as firmly as the one it was + created at. + """ + from superset.versioning.changes import ENTITY_KIND_BY_CLASS_NAME + + shadow = { + Slice: "slices_version", + Dashboard: "dashboards_version", + SqlaTable: "tables_version", + }[model] + kind = ENTITY_KIND_BY_CLASS_NAME[model.__name__] + # The transaction may be shared across entities — insert it once. + exists = db.session.execute( + sa.text("SELECT 1 FROM version_transaction WHERE id = :t"), {"t": tx_id} + ).first() + if not exists: + db.session.execute( + sa.text( + "INSERT INTO version_transaction (id, issued_at) VALUES (:t, :ts)" + ), + {"t": tx_id, "ts": datetime.utcnow()}, + ) + if ( + end_tx_id is not None + and not db.session.execute( + sa.text("SELECT 1 FROM version_transaction WHERE id = :t"), + {"t": end_tx_id}, + ).first() + ): + db.session.execute( + sa.text( + "INSERT INTO version_transaction (id, issued_at) VALUES (:t, :ts)" + ), + {"t": end_tx_id, "ts": datetime.utcnow()}, + ) + db.session.execute( + sa.text( + f"INSERT INTO {shadow} " # noqa: S608 + "(id, transaction_id, end_transaction_id, operation_type) " + "VALUES (:i, :t, :e, 0)" + ), + {"i": entity_id, "t": tx_id, "e": end_tx_id}, + ) + db.session.execute( + sa.text( + "INSERT INTO version_changes " + "(transaction_id, entity_kind, entity_id, sequence, kind, " + "operation, path) VALUES (:t, :k, :i, 1, 'set', 0, :p)" + ), + {"t": tx_id, "k": kind, "i": entity_id, "p": '["x"]'}, + ) + db.session.commit() + + # -- cleanup ------------------------------------------------------------ + + def _cleanup(self) -> None: + db.session.rollback() + from superset.connectors.sqla.models import RowLevelSecurityFilter + from superset.reports.models import ReportSchedule + + for report in db.session.query(ReportSchedule).filter( + ReportSchedule.name.like(f"{_PREFIX}%") + ): + db.session.delete(report) + for rule in db.session.query(RowLevelSecurityFilter).filter( + RowLevelSecurityFilter.name.like(f"{_PREFIX}%") + ): + db.session.delete(rule) + db.session.commit() + for model in (Dashboard, Slice, SqlaTable): + rows = ( + db.session.query(model) + .execution_options(**{SKIP_VISIBILITY_FILTER_CLASSES: {model}}) + .all() + ) + for row in rows: + name = ( + getattr(row, "slice_name", None) + or getattr(row, "dashboard_title", None) + or getattr(row, "table_name", "") + ) + if str(name).startswith(_PREFIX): + db.session.delete(row) + for d in db.session.query(Database).filter( + Database.database_name.like(f"{_PREFIX}%") + ): + db.session.delete(d) + db.session.commit() + # prefix-named tags + their tagged_object rows + from superset.tags.models import Tag, TaggedObject + + tag_ids = [ + t.id for t in db.session.query(Tag).filter(Tag.name.like(f"{_PREFIX}%")) + ] + if tag_ids: + db.session.query(TaggedObject).filter( + TaggedObject.tag_id.in_(tag_ids) + ).delete(synchronize_session=False) + db.session.query(Tag).filter(Tag.id.in_(tag_ids)).delete( + synchronize_session=False + ) + db.session.commit() + # Clear all version-capture rows (the test DB runs with capture ON, so + # creating/deleting entities above also writes shadow rows) and audit + # rows. Children before version_transaction for FK safety. + for tbl in ( + "dashboard_slices_version", + "slices_version", + "dashboards_version", + "tables_version", + "table_columns_version", + "sql_metrics_version", + "version_changes", + "version_transaction", + ): + db.session.execute(sa.text(f"DELETE FROM {tbl}")) # noqa: S608 + db.session.execute(sa.text("DELETE FROM purge_audit_log")) + db.session.commit() diff --git a/tests/integration_tests/deletion_retention/audit_tests.py b/tests/integration_tests/deletion_retention/audit_tests.py new file mode 100644 index 00000000000..7fe465e62c0 --- /dev/null +++ b/tests/integration_tests/deletion_retention/audit_tests.py @@ -0,0 +1,109 @@ +# 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 write-ahead purge audit record.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from unittest.mock import patch +from uuid import UUID + +from superset import db +from superset.commands.deletion_retention import audit +from superset.commands.deletion_retention.audit import PurgeAuditLog +from superset.models.slice import Slice +from superset.tasks.deletion_retention import _purge_impl + +from ._base import DeletionRetentionTestBase + + +class TestPurgeAudit(DeletionRetentionTestBase): + def test_write_ahead_then_confirm(self) -> None: + """A purge writes a pending audit row up front and flips it to + confirmed after the delete commits.""" + chart = self.make_chart("audited") + chart_uuid = str(chart.uuid) + self.soft_delete(chart, days_ago=90) + + _purge_impl(30, dry_run=False) + + row = db.session.query(PurgeAuditLog).filter_by(entity_uuid=chart_uuid).one() + assert row.status == audit.STATUS_CONFIRMED + assert row.trigger == audit.TRIGGER_RETENTION + assert row.actor == audit.ACTOR_SYSTEM + assert row.confirmed_on is not None + assert isinstance(row.id, UUID) + + def test_known_failure_finalizes_audit_row(self) -> None: + """A known cascade failure is durable but does not remain pending.""" + chart = self.make_chart("crash") + chart_id, chart_uuid = chart.id, str(chart.uuid) + self.soft_delete(chart, days_ago=90) + + # Make the cascade blow up after the write-ahead row is committed. + with patch( + "superset.tasks.deletion_retention.cascade_hard_delete", + side_effect=RuntimeError("boom"), + ): + result = _purge_impl(30, dry_run=False) + + # the run records the failure and does not purge + assert result["cascade_failures"] == 1 + assert self.exists(Slice, chart_id) + # the write-ahead row survives and is finalized as failed + row = db.session.query(PurgeAuditLog).filter_by(entity_uuid=chart_uuid).one() + assert row.status == audit.STATUS_FAILED + assert row.confirmed_on is None + + def test_reconcile_confirms_pending_after_entity_commit(self) -> None: + """A crash after entity commit is reconciled to confirmed.""" + chart = self.make_chart("committed_crash") + chart_uuid = str(chart.uuid) + self.soft_delete(chart, days_ago=90) + + with patch("superset.tasks.deletion_retention.audit.confirm"): + _purge_impl(30, dry_run=False) + + row = db.session.query(PurgeAuditLog).filter_by(entity_uuid=chart_uuid).one() + assert row.status == audit.STATUS_PENDING + + result = audit.reconcile_pending( + stale_before=datetime.utcnow() + timedelta(seconds=1) + ) + db.session.expire_all() + + assert result == {"reconciled": 1, "confirmed": 1, "failed": 0} + assert row.status == audit.STATUS_CONFIRMED + assert row.confirmed_on is not None + + def test_reconcile_fails_pending_when_entity_survives(self) -> None: + """A stale attempt with a surviving entity is closed as failed.""" + chart = self.make_chart("surviving_pending") + record_id = audit.write_ahead( + trigger=audit.TRIGGER_RETENTION, + actor=audit.ACTOR_SYSTEM, + entity_type="slices", + entity_uuid=str(chart.uuid), + ) + + result = audit.reconcile_pending( + stale_before=datetime.utcnow() + timedelta(seconds=1) + ) + db.session.expire_all() + + assert result == {"reconciled": 1, "confirmed": 0, "failed": 1} + assert db.session.get(PurgeAuditLog, record_id).status == audit.STATUS_FAILED diff --git a/tests/integration_tests/deletion_retention/force_purge_tests.py b/tests/integration_tests/deletion_retention/force_purge_tests.py new file mode 100644 index 00000000000..00abb5d5afe --- /dev/null +++ b/tests/integration_tests/deletion_retention/force_purge_tests.py @@ -0,0 +1,234 @@ +# 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 compliance force-purge.""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from superset import db +from superset.commands.deletion_retention.audit import PurgeAuditLog +from superset.commands.deletion_retention.force_purge import ( + AmbiguousPurgeTargetError, + ForcePurgeCommand, +) +from superset.connectors.sqla.models import SqlaTable +from superset.models.dashboard import Dashboard +from superset.models.slice import Slice +from superset.reports.models import ReportSchedule + +from ._base import DeletionRetentionTestBase + + +class TestForcePurge(DeletionRetentionTestBase): + def test_force_purge_live_entity(self) -> None: + """Force-purge removes a *live* entity (never + soft-deleted) immediately, ignoring the window.""" + chart = self.make_chart("live") + chart_id, chart_uuid = chart.id, str(chart.uuid) + + result = ForcePurgeCommand(chart_uuid).run() + + assert result["purged"] is True + assert result["entity_type"] == "chart" + assert not self.exists(Slice, chart_id) + + def test_force_purge_soft_deleted_entity_and_history(self) -> None: + """Force-purge a soft-deleted entity removes it and its + version history; the audit record survives.""" + chart = self.make_chart("softdel") + chart_id, chart_uuid = chart.id, str(chart.uuid) + self.forge_version_row(Slice, chart_id, tx_id=990050) + self.soft_delete(chart, days_ago=1) # inside the window; force ignores it + + ForcePurgeCommand(chart_uuid).run() + + assert not self.exists(Slice, chart_id) + assert ( + self.count( + "SELECT count(*) FROM slices_version WHERE id = :i", {"i": chart_id} + ) + == 0 + ) + audit = db.session.query(PurgeAuditLog).filter_by(entity_uuid=chart_uuid).all() + assert [(a.trigger, a.status) for a in audit] == [("force", "confirmed")] + + def test_force_purge_idempotent(self) -> None: + """Re-running force-purge on a gone UUID is a no-op.""" + chart = self.make_chart("once") + chart_uuid = str(chart.uuid) + ForcePurgeCommand(chart_uuid).run() + + result = ForcePurgeCommand(chart_uuid).run() + + assert result["purged"] is False + assert result["reason"] == "not_found" + + def test_force_purge_dataset_leaves_chart_dangling(self) -> None: + """Force-purging a dataset referenced by a live + chart succeeds, leaves the chart's datasource_id dangling (chart row + unchanged), and records the affected chart in the audit entry.""" + chart = self.make_chart("dep", dataset=self.dataset) + chart_id, chart_uuid = chart.id, str(chart.uuid) + ds_id, ds_uuid = self.dataset.id, str(self.dataset.uuid) + + result = ForcePurgeCommand(ds_uuid).run() + + assert result["purged"] is True + assert not self.exists(SqlaTable, ds_id) + kept = db.session.query(Slice).filter(Slice.id == chart_id).one() + assert kept.datasource_id == ds_id # dangling, unmodified + assert chart_uuid in result["dangling_chart_uuids"] + audit = db.session.query(PurgeAuditLog).filter_by(entity_uuid=ds_uuid).one() + assert audit.affected_referrers + assert chart_uuid in audit.affected_referrers + + def test_force_purge_counts_removed_dashboard_slices_before_db_cascade( + self, + ) -> None: + """The removed join count is accurate with FK enforcement enabled.""" + chart = self.make_chart("counted") + dashboard = self.make_dashboard("counted", slices=[chart]) + dashboard_id, chart_uuid = dashboard.id, str(chart.uuid) + + result = ForcePurgeCommand(chart_uuid).run() + + assert result["removed_dashboard_slices"] == 1 + assert self.exists(Dashboard, dashboard_id) + audit = db.session.query(PurgeAuditLog).filter_by(entity_uuid=chart_uuid).one() + assert audit.removed_dashboard_slices == 1 + + def test_force_purge_preserves_report_reference_blocker(self) -> None: + """Force bypasses age/state, not ordinary deletion restrictions.""" + chart = self.make_chart("force_reported") + report = ReportSchedule( + type="Report", + name="retention_it_force_report", + crontab="0 0 * * *", + chart=chart, + ) + db.session.add(report) + db.session.commit() + chart_id, chart_uuid = chart.id, str(chart.uuid) + + with patch( + "superset.commands.deletion_retention.force_purge.logger.info" + ) as log_info: + result = ForcePurgeCommand(chart_uuid).run() + + assert result["purged"] is False + assert result["reason"] == "blocked" + assert self.exists(Slice, chart_id) + row = db.session.query(PurgeAuditLog).filter_by(entity_uuid=chart_uuid).one() + assert row.status == "blocked" + log_info.assert_called_once_with( + "force_purge: blocked %s uuid=%s reason=%s", + "chart", + chart_uuid, + "associated alerts or reports exist", + ) + + def test_force_purge_refuses_an_ambiguous_uuid(self) -> None: + """A UUID matching two entity types is refused, not guessed. + + UUID uniqueness is per table, and the import APIs accept + caller-supplied UUIDs, so an operator's bare UUID can legitimately + match more than one row. Purging the first match found would let a + compliance deletion destroy an entity nobody asked about. + """ + chart = self.make_chart("ambiguous_chart") + dashboard = self.make_dashboard("ambiguous_dash") + shared = chart.uuid + dashboard.uuid = shared + db.session.commit() + chart_id, dashboard_id = chart.id, dashboard.id + self.soft_delete(chart, days_ago=90) + + with pytest.raises(AmbiguousPurgeTargetError): + ForcePurgeCommand(str(shared)).run() + + # Neither is touched. + assert self.exists(Slice, chart_id) + assert self.exists(Dashboard, dashboard_id) + + def test_force_purge_with_a_model_resolves_only_that_type(self) -> None: + """Given the type, the same ambiguous UUID purges exactly one row.""" + chart = self.make_chart("scoped_chart") + dashboard = self.make_dashboard("scoped_dash") + shared = chart.uuid + dashboard.uuid = shared + db.session.commit() + chart_id, dashboard_id = chart.id, dashboard.id + self.soft_delete(chart, days_ago=90) + + result = ForcePurgeCommand(str(shared), model_cls=Slice).run() + + assert result["purged"] is True + assert not self.exists(Slice, chart_id) + assert self.exists(Dashboard, dashboard_id) + + def test_cli_force_purge_reports_ambiguity_as_an_operator_error(self) -> None: + """The refusal reaches the operator as a message, not a traceback. + + It surfaces *after* the irreversible confirmation prompt has been + answered, which is the failure mode the ``type=click.UUID`` validation + on the same command exists to avoid. + """ + from click.testing import CliRunner + + from superset.cli.deletion_retention import force_purge + + chart = self.make_chart("cli_ambiguous_chart") + dashboard = self.make_dashboard("cli_ambiguous_dash") + shared = chart.uuid + dashboard.uuid = shared + db.session.commit() + chart_id, dashboard_id = chart.id, dashboard.id + self.soft_delete(chart, days_ago=90) + + result = CliRunner().invoke(force_purge, ["--uuid", str(shared), "--yes"]) + + assert result.exit_code != 0 + assert not isinstance(result.exception, AmbiguousPurgeTargetError) + assert "--type" in result.output + # The refusal is not a partial purge. + assert self.exists(Slice, chart_id) + assert self.exists(Dashboard, dashboard_id) + + def test_cli_force_purge_type_option_disambiguates(self) -> None: + """The escape hatch the error names actually exists and works.""" + from click.testing import CliRunner + + from superset.cli.deletion_retention import force_purge + + chart = self.make_chart("cli_typed_chart") + dashboard = self.make_dashboard("cli_typed_dash") + shared = chart.uuid + dashboard.uuid = shared + db.session.commit() + chart_id, dashboard_id = chart.id, dashboard.id + self.soft_delete(chart, days_ago=90) + + result = CliRunner().invoke( + force_purge, ["--uuid", str(shared), "--type", "chart", "--yes"] + ) + + assert result.exit_code == 0, result.output + assert not self.exists(Slice, chart_id) + assert self.exists(Dashboard, dashboard_id) diff --git a/tests/integration_tests/deletion_retention/purge_tests.py b/tests/integration_tests/deletion_retention/purge_tests.py new file mode 100644 index 00000000000..a1ffacb4457 --- /dev/null +++ b/tests/integration_tests/deletion_retention/purge_tests.py @@ -0,0 +1,631 @@ +# 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.orm.attributes import set_committed_value +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_does_not_finalize_pending_audit_rows(self) -> None: + """A dry run reports; it does not resolve another run's audit state. + + Reconciliation finalizes stale pending rows, which is a durable write. + An operator sizing up a rollout would otherwise alter the very record + they are inspecting. + """ + with patch( + "superset.tasks.deletion_retention.audit.reconcile_pending" + ) as reconcile: + _purge(window=30, dry_run=True) + assert reconcile.call_count == 0 + + # ...and a real run still reconciles. + _purge(window=30, dry_run=False) + assert reconcile.call_count == 1 + + 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 + dashboard = self.make_dashboard("fkoffdash", slices=[chart]) + dashboard_id = dashboard.id + # Embedded config is a delete-orphan child whose removal must not + # depend on the DB cascade either. + from superset.models.embedded_dashboard import EmbeddedDashboard + + db.session.add(EmbeddedDashboard(dashboard_id=dashboard_id)) + db.session.commit() + self.soft_delete(chart, days_ago=90) + self.soft_delete(dashboard, days_ago=90) + + db.session.execute(sa.text("PRAGMA foreign_keys=OFF")) + try: + _purge(window=30) + finally: + # The connection is pooled; a later test must not inherit + # disabled FK enforcement. + db.session.execute(sa.text("PRAGMA foreign_keys=ON")) + + assert not self.exists(Slice, chart_id) + assert ( + self.count( + "SELECT count(*) FROM dashboard_slices WHERE slice_id = :i", + {"i": chart_id}, + ) + == 0 + ) + assert ( + self.count( + "SELECT count(*) FROM embedded_dashboards WHERE dashboard_id = :i", + {"i": dashboard_id}, + ) + == 0 + ) + + def test_purge_writes_no_association_shadows_with_capture_on(self) -> None: + """A purge must not create association version shadows: the + Core deletes on dashboard_slices queue Continuum statements that the + suppression context discards before commit. After purging the + dashboard, no dashboard_slices_version rows for it remain — neither + pre-existing (its history is cascaded) nor purge-queued.""" + chart = self.make_chart("noshadow_chart") + dashboard = self.make_dashboard("noshadow_dash", slices=[chart]) + dashboard_id = dashboard.id + self.soft_delete(dashboard, days_ago=90) + + _purge(window=30) + + assert not self.exists(Dashboard, dashboard_id) + assert ( + self.count( + "SELECT count(*) FROM dashboard_slices_version WHERE dashboard_id = :i", + {"i": dashboard_id}, + ) + == 0 + ) + + def test_version_history_removed_and_shared_tx_preserved(self) -> None: + """A purged entity's version shadows and scoped + version_changes are removed and a sole-owner transaction swept, while a + transaction shared with a surviving entity is preserved.""" + purged = self.make_chart("hist_purged") + survivor = self.make_chart("hist_survivor") + purged_id, survivor_id = purged.id, survivor.id + # shared transaction owns shadow rows for both charts + self.forge_version_row(Slice, purged_id, tx_id=990001) + self.forge_version_row(Slice, survivor_id, tx_id=990001) + # sole-owner transaction for the purged chart + self.forge_version_row(Slice, purged_id, tx_id=990002) + self.soft_delete(purged, days_ago=90) + + _purge(window=30) + + # purged entity's history gone + assert ( + self.count( + "SELECT count(*) FROM slices_version WHERE id = :i", {"i": purged_id} + ) + == 0 + ) + assert ( + self.count( + "SELECT count(*) FROM version_changes WHERE entity_id = :i " + "AND entity_kind = 'chart'", + {"i": purged_id}, + ) + == 0 + ) + # sole-owner transaction swept; shared transaction kept (survivor) + assert ( + self.count("SELECT count(*) FROM version_transaction WHERE id = 990002", {}) + == 0 + ) + assert ( + self.count("SELECT count(*) FROM version_transaction WHERE id = 990001", {}) + == 1 + ) + # the survivor's shadow row on the shared transaction is preserved + # (capture-on may add a baseline row too, so assert on the forged one) + assert ( + self.count( + "SELECT count(*) FROM slices_version WHERE id = :i " + "AND transaction_id = 990001", + {"i": survivor_id}, + ) + == 1 + ) + + def test_transaction_closing_a_survivor_row_is_not_swept(self) -> None: + """A transaction referenced only through a survivor's + ``end_transaction_id`` is still referenced, so it must survive the + orphan sweep. + + A shadow row points at two transactions: the one that created it and, + once a later edit closes it, the one that ended it. Both are foreign + keys. Sweeping on the created-at side alone judges the closing + transaction orphaned while the survivor's row still points at it, and + the delete fails the foreign key — surfacing to the operator as + "blocked by existing deletion rules" rather than as the incomplete + cascade it is. + """ + purged = self.make_chart("endtx_purged") + survivor = self.make_chart("endtx_survivor") + purged_id, survivor_id = purged.id, survivor.id + # The purged chart is the only entity *created* at 990004. + self.forge_version_row(Slice, purged_id, tx_id=990004) + # The survivor's earlier row was *closed* at that same transaction, so + # nothing surviving references 990004 through transaction_id alone. + self.forge_version_row(Slice, survivor_id, tx_id=990003, end_tx_id=990004) + self.soft_delete(purged, days_ago=90) + + _purge(window=30) + + # The purge completed rather than reporting itself blocked. + assert not self.exists(Slice, purged_id) + # The closing transaction is retained because the survivor still + # points at it. + assert ( + self.count("SELECT count(*) FROM version_transaction WHERE id = 990004", {}) + == 1 + ) + assert ( + self.count( + "SELECT count(*) FROM slices_version WHERE id = :i " + "AND end_transaction_id = 990004", + {"i": survivor_id}, + ) + == 1 + ) + + def test_version_tables_absent_noop(self) -> None: + """When the version tables are absent the + version cascade no-ops cleanly and the entity is still purged.""" + chart = self.make_chart("noversion") + chart_id = chart.id + self.soft_delete(chart, days_ago=90) + + with patch( + "superset.commands.deletion_retention.purge_cascade." + "_version_tables_present", + return_value=False, + ): + result = _purge(window=30) + + assert result["purged"].get("slices") == 1 + assert not self.exists(Slice, chart_id) + + +class TestPurgeIdentityGuard(DeletionRetentionTestBase): + """The audit row must name the entity that was actually purged.""" + + def test_identity_drift_between_audit_and_purge_skips_the_entity(self) -> None: + """A row whose identity changed after the audit write is not purged. + + ``_purge_one`` snapshots the uuid, writes the write-ahead audit row, + then re-reads the entity by id. Ids can be recycled -- SQLite reuses + rowids -- so the row under that id may no longer be the one the audit + describes. The cascade's conditional claim would still refuse to + destroy anything ineligible, making this an attribution guard rather + than a destructive one: an audit row identifying the wrong object is + worse than a skipped purge. + """ + chart = self.make_chart("identity_drift") + chart_id = chart.id + self.soft_delete(chart, days_ago=90) + + real_uuid = str(chart.uuid) + # First call feeds the audit row; the second is the post-audit + # re-check, where the identity is made to differ. + uuids = iter([real_uuid, "00000000-0000-0000-0000-0000deadbeef"]) + + with ( + patch( + "superset.tasks.deletion_retention.entity_uuid", + side_effect=lambda _entity: next(uuids), + ), + patch("superset.tasks.deletion_retention.cascade_hard_delete") as cascade, + ): + stats = _purge() + + cascade.assert_not_called() + assert self.exists(Slice, chart_id) + assert stats["purged"] == {} + + row = ( + db.session.query(audit.PurgeAuditLog) + .filter(audit.PurgeAuditLog.entity_uuid == real_uuid) + .one() + ) + assert row.status == audit.STATUS_FAILED + + def test_cascade_refuses_a_row_whose_uuid_no_longer_matches(self) -> None: + """A reused id does not let the cascade purge a stranger. + + The id alone is not an identity: callers snapshot the entity well + before the cascade runs -- the retention task writes an audit row in + between -- and an id freed and reissued in that gap would otherwise be + purged under the snapshot's name. Re-checking the uuid before the lock + narrows that window; the predicate on the locked claim closes it. + """ + chart = self.make_chart("uuid_drift") + chart_id = chart.id + self.soft_delete(chart, days_ago=90) + snapshot_uuid = str(chart.uuid) + + # Stand in for the id being reissued: the stored row is now a different + # entity, while the caller still holds the snapshot it resolved. + db.session.execute( + sa.update(Slice.__table__) + .where(Slice.__table__.c.id == chart_id) + .values(uuid="00000000-0000-0000-0000-00000000beef") + ) + db.session.commit() + db.session.refresh(chart) + # Restore the snapshot's view without marking the attribute dirty, so + # the cascade sees the uuid its caller resolved rather than the row's. + set_committed_value(chart, "uuid", snapshot_uuid) + + result = cascade_hard_delete( + db.session, + chart, + enforce_window=True, + cutoff=datetime.now() - timedelta(days=30), + ) + db.session.commit() + + assert result.purged is False + assert self.exists(Slice, chart_id) diff --git a/tests/integration_tests/deletion_retention/window_tests.py b/tests/integration_tests/deletion_retention/window_tests.py new file mode 100644 index 00000000000..2eb6d2260bc --- /dev/null +++ b/tests/integration_tests/deletion_retention/window_tests.py @@ -0,0 +1,99 @@ +# 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 per-workspace retention window.""" + +from __future__ import annotations + +from unittest.mock import patch + +from flask import current_app + +from superset.commands.deletion_retention.window import resolve_retention_window +from superset.key_value.shared_entries import get_shared_value, upsert_shared_value +from superset.key_value.types import SharedKey +from superset.models.slice import Slice +from superset.tasks.deletion_retention import _purge_impl, purge_soft_deleted + +from ._base import DeletionRetentionTestBase + + +class TestRetentionWindow(DeletionRetentionTestBase): + def tearDown(self) -> None: + # clear any shared window value this test set so the env default is + # restored for other tests + from uuid import uuid3 + + from superset import db + from superset.daos.key_value import KeyValueDAO + from superset.key_value.shared_entries import RESOURCE + from superset.key_value.utils import get_uuid_namespace + + try: + KeyValueDAO.delete_entry( + RESOURCE, + uuid3(get_uuid_namespace(""), SharedKey.SOFT_DELETE_RETENTION_DAYS), + ) + db.session.commit() + except Exception: # pylint: disable=broad-except + db.session.rollback() + super().tearDown() + + def test_shared_value_overrides_env_and_is_used_by_task(self) -> None: + """A per-workspace shared value takes + precedence over the env default and is honored by the purge.""" + upsert_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS, 10) + assert resolve_retention_window() == 10 + + # 20 days old: still inside the env default (30) but past the 10-day + # per-workspace override, so the override is what gets it purged + chart = self.make_chart("c") + chart_id = chart.id + self.soft_delete(chart, days_ago=20) + + with ( + patch( + "superset.tasks.deletion_retention.feature_flag_manager." + "is_feature_enabled", + return_value=True, + ), + patch.dict( + current_app.config, + {"SOFT_DELETE_PURGE_DRY_RUN": False}, + ), + ): + result: dict[str, object] = purge_soft_deleted.run() + + assert result["purged"] == {"slices": 1} + assert not self.exists(Slice, chart_id) + + def test_upsert_is_idempotent(self) -> None: + """Re-setting the window via upsert does not raise and keeps the + latest value (the CLI uses upsert, not the non-idempotent set).""" + upsert_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS, 15) + upsert_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS, 20) + assert get_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS) == 20 + assert resolve_retention_window() == 20 + + def test_zero_disables(self) -> None: + """A zero shared value disables the time-based purge.""" + upsert_shared_value(SharedKey.SOFT_DELETE_RETENTION_DAYS, 0) + chart = self.make_chart("c") + chart_id = chart.id + self.soft_delete(chart, days_ago=90) + + assert _purge_impl(resolve_retention_window(), dry_run=False) == {"skipped": 1} + assert self.exists(Slice, chart_id) diff --git a/tests/unit_tests/tasks/test_deletion_retention.py b/tests/unit_tests/tasks/test_deletion_retention.py new file mode 100644 index 00000000000..3615cb2a2e0 --- /dev/null +++ b/tests/unit_tests/tasks/test_deletion_retention.py @@ -0,0 +1,222 @@ +# 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. +"""Unit tests for deletion-retention configuration and window resolution. + +The shared value overrides config, an unset value falls back to config, ``0`` +is preserved as the disable value, and malformed shared values use the fallback. +""" + +import runpy +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from flask.config import Config + + +@pytest.fixture +def app_config(app_context: None) -> Config: + from flask import current_app + + current_app.config["SOFT_DELETE_RETENTION_DAYS"] = 30 + return current_app.config + + +def _resolve() -> int: + from superset.commands.deletion_retention.window import resolve_retention_window + + return resolve_retention_window() + + +def test_unset_falls_back_to_config(app_config: Config) -> None: + with patch( + "superset.commands.deletion_retention.window.get_shared_value", + return_value=None, + ): + assert _resolve() == 30 + + +def test_shared_value_overrides_config(app_config: Config) -> None: + with patch( + "superset.commands.deletion_retention.window.get_shared_value", + return_value=7, + ): + assert _resolve() == 7 + + +def test_zero_shared_value_is_preserved_not_coerced(app_config: Config) -> None: + # `0` is a meaningful "disable"; it must survive (never `or`-coerced to 30). + with patch( + "superset.commands.deletion_retention.window.get_shared_value", + return_value=0, + ): + assert _resolve() == 0 + + +def test_malformed_shared_value_falls_back(app_config: Config) -> None: + for bad in ("oops", -3, True, 1.5): + with patch( + "superset.commands.deletion_retention.window.get_shared_value", + return_value=bad, + ): + assert _resolve() == 30 + + +@pytest.mark.parametrize("configured", ["oops", -3, True, None]) +def test_malformed_config_value_falls_back( + app_config: Config, configured: object +) -> None: + app_config["SOFT_DELETE_RETENTION_DAYS"] = configured + with patch( + "superset.commands.deletion_retention.window.get_shared_value", + return_value=None, + ): + assert _resolve() == 30 + + +def test_window_zero_disables_the_task(app_context: None) -> None: + # A zero window short-circuits the purge entirely. + import superset.tasks.deletion_retention as mod + + with patch.object(mod, "_soft_delete_models") as models: + result = mod._purge_impl(0, dry_run=False) + assert result == {"skipped": 1} + models.assert_not_called() + + +def test_clock_uses_now_not_utcnow() -> None: + import superset.tasks.deletion_retention as mod + from superset.models.slice import Slice + + now = datetime(2026, 7, 13, 12, 0) + with ( + patch.object(mod, "datetime") as clock, + patch.object(mod, "_soft_delete_models", return_value=[Slice]), + patch.object(mod, "_purge_model", return_value=(0, 0, 0, 0)) as purge, + patch.object(mod.audit, "reconcile_pending"), + ): + clock.now.return_value = now + mod._purge_impl(30, dry_run=False) + + purge.assert_called_once_with(Slice, now - timedelta(days=30), False) + + +def test_default_config_is_safe() -> None: + from superset import config + + assert config.SOFT_DELETE_RETENTION_DAYS == 30 + assert config.SOFT_DELETE_PURGE_DRY_RUN is True + + +def test_default_celery_config_registers_daily_purge() -> None: + from superset import config + + assert "superset.tasks.deletion_retention" in config.CeleryConfig.imports + entry: dict[str, Any] = config.CeleryConfig.beat_schedule[ + "deletion_retention.purge_soft_deleted" + ] + assert entry["task"] == "deletion_retention.purge_soft_deleted" + assert entry["schedule"].minute == {0} + assert entry["schedule"].hour == {0} + + +def test_docker_celery_config_registers_daily_purge() -> None: + config_path = Path(__file__).parents[3] / "docker/pythonpath_dev/superset_config.py" + with patch("flask_caching.backends.filesystemcache.FileSystemCache"): + docker_config: dict[str, Any] = runpy.run_path(str(config_path)) + celery_config: type[Any] = docker_config["CeleryConfig"] + + assert "superset.tasks.deletion_retention" in celery_config.imports + entry: dict[str, Any] = celery_config.beat_schedule[ + "deletion_retention.purge_soft_deleted" + ] + assert entry["task"] == "deletion_retention.purge_soft_deleted" + assert entry["schedule"].minute == {0} + assert entry["schedule"].hour == {0} + + +def test_purge_suppression_is_session_scoped() -> None: + from superset.commands.deletion_retention.purge_cascade import ( + suppress_purge_association_versions, + ) + + existing = object() + purge_statement = object() + unit_of_work = MagicMock() + unit_of_work.pending_statements = [existing] + manager = MagicMock() + manager.options = {"versioning": True, "native_versioning": False} + manager.unit_of_work.return_value = unit_of_work + session = MagicMock() + + with patch("sqlalchemy_continuum.versioning_manager", manager): + with suppress_purge_association_versions(session): + assert manager.options["versioning"] is True + unit_of_work.pending_statements.append(purge_statement) + + assert unit_of_work.pending_statements == [existing] + manager.unit_of_work.assert_called_once_with(session) + + +def test_purge_model_counts_only_committed_deletions(app_context: None) -> None: + import superset.tasks.deletion_retention as mod + from superset.commands.deletion_retention.purge_cascade import CascadeResult + from superset.models.slice import Slice + + lost_race: CascadeResult = CascadeResult( + purged=False, entity_type="chart", entity_uuid="lost-race" + ) + with ( + patch.object(mod, "_iter_eligible_ids", return_value=[[1]]), + patch.object(mod, "_purge_one", return_value=lost_race), + ): + result: tuple[int, int, int, int] = mod._purge_model( + Slice, datetime.now(), dry_run=False + ) + + assert result == (0, 0, 0, 0) + + +def test_scheduled_purge_fails_closed_when_write_ahead_fails( + app_context: None, +) -> None: + """An unauditable scheduled purge must not delete: the entity is + skipped (counted as a failure) and retried next run.""" + import superset.tasks.deletion_retention as mod + from superset.models.slice import Slice + + entity = MagicMock(id=1) + with ( + patch.object(mod, "_iter_eligible_ids", return_value=[[1]]), + patch.object(mod, "skip_visibility_filter"), + patch.object(mod, "entity_uuid", return_value="u-1"), + patch.object(mod, "dashboard_slice_count", return_value=0), + patch.object(mod, "cascade_hard_delete") as cascade, + patch.object(mod.db, "session") as session, + patch.object(mod.audit, "write_ahead", return_value=None), + ): + session.get.return_value = entity + result: tuple[int, int, int, int] = mod._purge_model( + Slice, datetime.now(), dry_run=False + ) + + cascade.assert_not_called() + purged, would, failures, blocked = result + assert (purged, would, blocked) == (0, 0, 0) + assert failures == 1