fix(deletion-retention): dedupe repeated blocked audits

UPDATING.md

@@ -486,7 +486,7 @@ Purging is **live by default** (`SOFT_DELETE_PURGE_DRY_RUN=False`), so the reten
 
 Deployments that replace the default `CELERY_CONFIG` must ensure workers register `superset.tasks.deletion_retention` and schedule the `deletion_retention.purge_soft_deleted` task themselves. The shipped Docker development config uses `imports` and includes both entries. While `SOFT_DELETE` is statically enabled, a missing beat entry logs a startup warning; when the override explicitly defines `imports`, a missing purge module is also reported.
 
-Operators can immediately erase a specific entity for compliance (GDPR) via `superset deletion-retention force-purge --uuid <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.
+Operators can immediately erase a specific entity for compliance (GDPR) via `superset deletion-retention force-purge --uuid <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 scheduled evaluation writes a provisional, content-free record to the new `purge_audit_log` table before the cascade starts. Meaningful retained outcomes survive the entity they name. Consecutive scheduled evaluations with the same blocked outcome suppress only the redundant current provisional record; completed outcomes, outcome transitions, and every force-purge attempt remain independent and immutable. The **scheduled** purge fails closed when its provisional record cannot be written, while **force-purge** proceeds even if the audit write fails — the operator is present and deletion outranks audit for a compliance erasure. Operators can monitor `deletion_retention.blocked_audit_suppressed` and `deletion_retention.blocked_audit_dedupe_fallback` to verify suppression and fail-safe fallback behavior without changing the existing blocked-workload gauge.
 
 ### Recently Archived view and permanent delete (purge) endpoints
 

superset/commands/deletion_retention/audit.py

@@ -14,16 +14,18 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Write-ahead purge audit record.
+"""Write-ahead purge audit records and retained purge outcomes.
 
-Every purge — time-based or force — writes an immutable record that
-**survives** the entity it names, on a **dedicated session** outside the
+Every purge evaluation — scheduled or force — writes a provisional record 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).
+next run. Completed records are immutable. Consecutive scheduled evaluations
+that remain blocked may discard only their current redundant provisional row;
+force-purge and other meaningful outcomes are retained independently.
 
 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
@@ -39,11 +41,13 @@
 from __future__ import annotations
 
 import logging
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import Any, cast
+from typing import Any, cast, Literal, TypeAlias
 from uuid import UUID
 
 import sqlalchemy as sa
+from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session, sessionmaker
 
 from superset import db
@@ -73,6 +77,17 @@ def _dedicated_session() -> Session:
 
 ACTOR_SYSTEM = "system"
 
+RetentionBlockedDisposition: TypeAlias = Literal["retained", "suppressed", "fallback"]
+
+
+@dataclass(frozen=True)
+class _AuditRecoverySnapshot:
+    id: UUID
+    actor: str
+    entity_type: str
+    entity_uuid: str | None
+    created_on: datetime
+
 
 def _utc_now() -> datetime:
     """Naive UTC now for the audit columns.
@@ -182,6 +197,146 @@ def block(record_id: UUID | None) -> None:
     finalize(record_id, STATUS_BLOCKED)
 
 
+def _capture_recovery_snapshot(record: PurgeAuditLog) -> _AuditRecoverySnapshot:
+    """Capture the content-free fields needed for fail-safe recovery."""
+    return _AuditRecoverySnapshot(
+        id=cast(UUID, record.id),
+        actor=str(record.actor),
+        entity_type=str(record.entity_type),
+        entity_uuid=record.entity_uuid,
+        created_on=cast(datetime, record.created_on),
+    )
+
+
+def _retention_predecessor(
+    session: Session, current: PurgeAuditLog
+) -> PurgeAuditLog | None:
+    """Return the latest row that could unambiguously precede ``current``."""
+    return session.execute(
+        sa.select(PurgeAuditLog)
+        .where(PurgeAuditLog.entity_uuid == current.entity_uuid)
+        .where(PurgeAuditLog.entity_type == current.entity_type)
+        .where(PurgeAuditLog.trigger == TRIGGER_RETENTION)
+        .where(PurgeAuditLog.created_on <= current.created_on)
+        .where(PurgeAuditLog.id != current.id)
+        .order_by(PurgeAuditLog.created_on.desc())
+        .limit(1)
+    ).scalar_one_or_none()
+
+
+def _suppress_redundant_block(
+    session: Session, current: PurgeAuditLog, predecessor: PurgeAuditLog | None
+) -> bool:
+    """Delete only a pending row with a strictly older blocked predecessor."""
+    if (
+        predecessor is None
+        or predecessor.created_on >= current.created_on
+        or predecessor.status != STATUS_BLOCKED
+    ):
+        return False
+    deleted_rows: int | None = session.execute(
+        sa.delete(PurgeAuditLog.__table__).where(
+            PurgeAuditLog.__table__.c.id == current.id,
+            PurgeAuditLog.__table__.c.status == STATUS_PENDING,
+        )
+    ).rowcount
+    if deleted_rows not in (0, 1):
+        raise SQLAlchemyError(
+            f"indeterminate purge audit suppression rowcount: {deleted_rows}"
+        )
+    return deleted_rows == 1
+
+
+def _retain_blocked(session: Session, record_id: UUID) -> None:
+    """Conditionally retain the current provisional row as blocked."""
+    session.execute(
+        sa.update(PurgeAuditLog.__table__)
+        .where(
+            PurgeAuditLog.__table__.c.id == record_id,
+            PurgeAuditLog.__table__.c.status == STATUS_PENDING,
+        )
+        .values(status=STATUS_BLOCKED, removed_dashboard_slices=0)
+    )
+
+
+def _recover_retention_blocked(
+    record_id: UUID, snapshot: _AuditRecoverySnapshot | None
+) -> RetentionBlockedDisposition:
+    """Retain blocked evidence on a fresh session after persistence uncertainty."""
+    recovery_session: Session = _dedicated_session()
+    try:
+        current: PurgeAuditLog | None = recovery_session.get(PurgeAuditLog, record_id)
+        if current is not None:
+            if current.status == STATUS_PENDING:
+                _retain_blocked(recovery_session, record_id)
+                recovery_session.commit()
+            return "fallback"
+        if snapshot is None:
+            return "fallback"
+        recovery_session.add(
+            PurgeAuditLog(
+                id=snapshot.id,
+                status=STATUS_BLOCKED,
+                trigger=TRIGGER_RETENTION,
+                actor=snapshot.actor,
+                entity_type=snapshot.entity_type,
+                entity_uuid=snapshot.entity_uuid,
+                removed_dashboard_slices=0,
+                created_on=snapshot.created_on,
+            )
+        )
+        recovery_session.commit()
+    except SQLAlchemyError:
+        recovery_session.rollback()
+        logger.warning(
+            "deletion_retention: failed to recover blocked audit row %s "
+            "entity_type=%s entity_uuid=%s",
+            record_id,
+            snapshot.entity_type if snapshot else None,
+            snapshot.entity_uuid if snapshot else None,
+            exc_info=True,
+        )
+    finally:
+        recovery_session.close()
+    return "fallback"
+
+
+def finalize_retention_blocked(
+    record_id: UUID | None,
+) -> RetentionBlockedDisposition:
+    """Finalize a scheduled blocker, suppressing only proven redundant evidence."""
+    if record_id is None:
+        return "fallback"
+    session: Session = _dedicated_session()
+    snapshot: _AuditRecoverySnapshot | None = None
+    try:
+        current: PurgeAuditLog | None = session.get(PurgeAuditLog, record_id)
+        if current is None:
+            return "fallback"
+        snapshot = _capture_recovery_snapshot(current)
+        if current.status != STATUS_PENDING or current.trigger != TRIGGER_RETENTION:
+            return "retained"
+        predecessor: PurgeAuditLog | None = None
+        if current.entity_uuid is not None:
+            predecessor = _retention_predecessor(session, current)
+        suppressed: bool = _suppress_redundant_block(session, current, predecessor)
+        if not suppressed:
+            _retain_blocked(session, record_id)
+        session.commit()
+        return "suppressed" if suppressed else "retained"
+    except SQLAlchemyError:
+        session.rollback()
+        logger.warning(
+            "deletion_retention: persistence uncertainty finalizing blocked "
+            "audit row %s",
+            record_id,
+            exc_info=True,
+        )
+    finally:
+        session.close()
+    return _recover_retention_blocked(record_id, snapshot)
+
+
 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

superset/migrations/versions/2026-08-06_18-00_b8d2f4a6c901_index_purge_audit_predecessor.py

@@ -0,0 +1,43 @@
+# 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.
+"""Index purge audit predecessor lookups.
+
+Revision ID: b8d2f4a6c901
+Revises: c4a1b8e2d739
+Create Date: 2026-08-06 18:00:00.000000
+
+"""
+
+from superset.migrations.shared.utils import create_index, drop_index
+
+# revision identifiers, used by Alembic.
+revision: str = "b8d2f4a6c901"
+down_revision: str = "c4a1b8e2d739"
+
+_INDEX_NAME: str = "ix_purge_audit_log_retention_predecessor"
+
+
+def upgrade() -> None:
+    create_index(
+        "purge_audit_log",
+        _INDEX_NAME,
+        ["entity_uuid", "entity_type", "trigger", "created_on"],
+    )
+
+
+def downgrade() -> None:
+    drop_index("purge_audit_log", _INDEX_NAME)

superset/models/purge_audit_log.py

@@ -52,6 +52,13 @@ class PurgeAuditLog(Model):
         # 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"),
+        sa.Index(
+            "ix_purge_audit_log_retention_predecessor",
+            "entity_uuid",
+            "entity_type",
+            "trigger",
+            "created_on",
+        ),
     )
 
     id = Column(UUIDType(binary=True), primary_key=True, default=uuid4)

superset/tasks/deletion_retention.py

@@ -280,7 +280,17 @@ def _purge_one(
             removed_dashboard_slices=result.removed_dashboard_slices,
         )
     elif result.blocked_reason is not None:
-        audit.block(record_id)
+        disposition: audit.RetentionBlockedDisposition = (
+            audit.finalize_retention_blocked(record_id)
+        )
+        if disposition == "suppressed":
+            stats_logger_manager.instance.incr(
+                f"{_METRIC_PREFIX}.blocked_audit_suppressed"
+            )
+        elif disposition == "fallback":
+            stats_logger_manager.instance.incr(
+                f"{_METRIC_PREFIX}.blocked_audit_dedupe_fallback"
+            )
     else:
         audit.fail(record_id)
     return result