sadpandajoe commented on code in PR #43490: URL: https://github.com/apache/superset/pull/43490#discussion_r3992450606
########## superset/commands/deletion_retention/prune_audit.py: ########## @@ -0,0 +1,726 @@ +# 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-only pruning of the ``purge_audit_log`` table. + +Bounds the audit history's growth without ever weakening its evidentiary +value. Three delete categories, applied in priority order under one shared +per-run batch budget: + +1. **Blocked duplicates** — within an entity's *current* blockage streak + (its ``blocked`` rows newer than the entity's newest streak-breaking + row), only the first row of each run of consecutive same-reason rows + survives: the streak's earliest row and the first row after every + change of block reason. Later same-reason repeats are removed + regardless of age. The survivors carry the "blocked since" fact and + the reason history — for coded reasons, the same rows the audit writer's + own suppression rule retains (reason-less legacy runs are additionally + collapsed to their earliest here) — and are never deleted while the + streak is current. +2. **Operational expiry** — ``blocked`` rows of *resolved* streaks and + ``failed`` rows older than ``PURGE_AUDIT_OPERATIONAL_RETENTION_DAYS`` + age out. +3. **Evidence expiry** — ``confirmed`` / ``target_absent`` rows are + untouchable unless ``PURGE_AUDIT_EVIDENCE_RETENTION_DAYS`` is + explicitly set (the operator's compliance assertion), and then only + rows older than that window. + +``pending`` rows belong to :func:`audit.reconcile_pending` and rows with +future timestamps (clock skew) are excluded from every category *and* from +streak classification, so a skewed writer cannot reclassify a live streak. + +Three invariants keep the survivor safe without a distributed lock, which +matters because runs can overlap and both ``reconcile_pending`` and the +purge path finalize rows concurrently. A duplicate is only ever deleted +when no concurrent transition could turn it into a survivor first: + +* **A boundary is never removed while it still bounds anything.** Evidence + expiry refuses to delete a row while an older ``blocked`` or ``pending`` + row for the same entity survives. Boundaries therefore never *recede*, + which would otherwise promote resolved rows into a current streak. +* **A blocked row whose classification is unstable is never deleted.** + Finalizing a ``pending`` row resolves it *in place*, keeping its original + timestamp, so an unresolved attempt is a boundary that may appear + mid-history at any moment — and the blocked row after it would become its + streak's survivor. Blocked rows preceded by an unresolved attempt are + therefore skipped by **both** the duplicate and the operational category. + Age does not stabilize the classification: an old blocked row inside a + live streak is exactly the "blocked for years" case FR-009 protects. + Without this, a boundary moving *forward* would demote the current + survivor and promote the next row into its place — possibly a row already + selected for deletion. +* **Deletes are conditional and counted from rowcounts.** A row whose + status changed since selection is not matched, so overlapping runs can + neither double-remove nor double-report. + +Boundaries moving forward past *all* of an entity's blocked rows leave no +survivor to promote, so a selected duplicate may be removed slightly ahead +of its retention window in that case. It was redundant either way and the +streak's earliest row is untouched. + +Timestamps order rows. Ties are possible — legacy second-precision rows, +two writers within one clock tick — and are resolved on the preserving +side: a ``pending`` row tied with a blocked row counts as preceding it (the +block is deferred until the attempt resolves); a blocked row tied with a +boundary sits on the boundary's *resolved* side (it ages out instead of +seeding a new current streak, and the boundary is not removed before it); +and tied same-reason blocked rows are all retained. + +Candidate selection is embedded in each ``DELETE`` statement. The derived-table +wrapper keeps that shape legal on MySQL while ensuring that a pending or +recovered row committed before the delete is evaluated participates in the +survivor and boundary predicates. No stale list of candidate ids crosses a +transaction boundary. + +Audit creation/recovery and every pruning batch take the same singleton +database write lock before assigning a timestamp or evaluating candidates. +The lock is held through commit, so an audit row cannot become visible in an +already-processed logical past. Automatic pruning still ships disabled by +default so operators explicitly choose their retention policy. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from functools import partial +from typing import Any, Literal, NamedTuple, TypeAlias + +import sqlalchemy as sa +from flask import current_app + +from superset import db +from superset.commands.deletion_retention.audit import ( + acquire_coordination_lock, + TRIGGER_FORCE, + utc_now, +) +from superset.models.purge_audit_log import ( + PurgeAuditLog, + STATUS_BLOCKED, + STATUS_CONFIRMED, + STATUS_FAILED, + STATUS_PENDING, + STATUS_TARGET_ABSENT, +) + +logger: logging.Logger = logging.getLogger(__name__) + +#: Operational records: noise-prone outcomes whose compliance value decays — +#: a blocked or failed purge leaves the object in place (FR-001). +OPERATIONAL_STATUSES: frozenset[str] = frozenset({STATUS_BLOCKED, STATUS_FAILED}) +#: Protected evidence: the only surviving trace of a destroyed object. +PROTECTED_STATUSES: frozenset[str] = frozenset({STATUS_CONFIRMED, STATUS_TARGET_ABSENT}) +#: The outcomes that end a blockage streak — proof the object is gone. +#: +#: ``failed`` is deliberately absent. A failed purge is an infrastructure +#: outcome (the cascade raised), not evidence the blockage cleared: the +#: policy that blocked the entity is untouched, so the blockage continues +#: across it. Treating ``failed`` as a boundary would let one transient +#: error demote the "blocked since" survivor to an ageing duplicate and +#: restate the blockage as beginning after the failure — losing exactly the +#: fact FR-003 exists to preserve. ``pending`` is provisional and likewise +#: neither joins nor breaks streaks. +_STREAK_BREAKING_STATUSES: frozenset[str] = frozenset( + {STATUS_CONFIRMED, STATUS_TARGET_ABSENT} +) + +#: Rows deleted per statement, matching the purge task's batch convention. +BATCH_SIZE: int = 500 +#: One shared budget for the whole run across all three categories; the +#: remaining backlog carries over to the next scheduled run (FR-004/SC-004). +MAX_BATCHES_PER_RUN: int = 10 + +OPERATIONAL_RETENTION_KEY: str = "PURGE_AUDIT_OPERATIONAL_RETENTION_DAYS" +EVIDENCE_RETENTION_KEY: str = "PURGE_AUDIT_EVIDENCE_RETENTION_DAYS" + + +class ResolvedWindow(NamedTuple): + """A retention window resolved from config. + + Distinguishes the three outcomes a caller must tell apart: a usable + window, the deliberate "off" default, and operator error. Collapsing + the last two into a bare ``None`` would force every caller to re-read + config to find out which it got. + """ + + days: int | None + invalid_key: str | None = None + + +def _validated_window(key: str, value: Any) -> ResolvedWindow: + """Validate a configured day count, failing closed on anything odd. + + An invalid value disables its category for the run rather than widening + removal (FR-005/SC-005). + """ + # bool is an int subclass and floats would silently truncate — both are + # config mistakes, not day counts, on a knob that deletes rows. + if not isinstance(value, bool) and not isinstance(value, float): + try: + days: int = int(value) + except (TypeError, ValueError): + days = 0 + if days > 0: + return ResolvedWindow(days) + logger.warning( + "prune_audit: invalid %s=%r; skipping this category for the run " + "(pruning never widens on bad configuration)", + key, + value, + ) + return ResolvedWindow(None, key) + + +def resolve_operational_retention_days() -> ResolvedWindow: + """The operational retention window, or a disabled window when invalid.""" + return _validated_window( + OPERATIONAL_RETENTION_KEY, current_app.config.get(OPERATIONAL_RETENTION_KEY) + ) + + +def resolve_evidence_retention_days() -> ResolvedWindow: + """The evidence expiration window; disabled unless explicitly opted in. + + Unset is the documented "never expire evidence" default (FR-006), not an + error, so it produces a disabled window with no warning. + """ + value: Any = current_app.config.get(EVIDENCE_RETENTION_KEY) + if value is None: + return ResolvedWindow(None) + return _validated_window(EVIDENCE_RETENTION_KEY, value) + + +@dataclass +class PruneRunResult: + """Per-category removal counts and run disposition for one pruning run.""" + + blocked_duplicates: int = 0 + operational_expired: int = 0 + evidence_expired: int = 0 + #: True when the shared batch budget ran out before every category's + #: candidates were drained; the remainder converges on later runs. + carried_over: bool = False + invalid_config_keys: list[str] = field(default_factory=list) + + @property + def total_removed(self) -> int: + """Total rows removed across every category.""" + return ( + self.blocked_duplicates + self.operational_expired + self.evidence_expired + ) + + def as_dict(self) -> dict[str, Any]: + """The task-return / log-line shape of this result.""" + return { + "removed": { + "blocked_duplicates": self.blocked_duplicates, + "operational_expired": self.operational_expired, + "evidence_expired": self.evidence_expired, + }, + "carried_over": self.carried_over, + "invalid_config_keys": list(self.invalid_config_keys), + } + + +def _streak_boundary_subquery(now: datetime) -> sa.Subquery: + """Per entity, the ``created_on`` of its newest streak-breaking row. + + Entities absent from this subquery have never been proven destroyed — + all their blocked rows form one current streak. Future-dated rows are + excluded so a skewed writer clock cannot push the boundary ahead of a + live streak and make its rows look resolved. + """ + table: sa.Table = PurgeAuditLog.__table__ + return ( + sa.select( + table.c.entity_type.label("entity_type"), + table.c.entity_uuid.label("entity_uuid"), + sa.func.max(table.c.created_on).label("boundary"), + ) + .where(table.c.status.in_(_STREAK_BREAKING_STATUSES)) + .where(table.c.entity_uuid.is_not(None)) + .where(table.c.created_on <= now) + .group_by(table.c.entity_type, table.c.entity_uuid) + .subquery("streak_boundary") + ) + + +def _with_boundary(table: sa.Table, boundary: sa.Subquery) -> sa.Join: + """Outer-join each row to its entity's streak boundary (NULL if none).""" + return table.outerjoin( + boundary, + sa.and_( + table.c.entity_type == boundary.c.entity_type, + table.c.entity_uuid == boundary.c.entity_uuid, + ), + ) + + +def _in_current_streak( + row: sa.FromClause, boundary: sa.Subquery +) -> sa.ColumnElement[bool]: + """Whether ``row`` is newer than its entity's boundary (or there is none). + + Strictly newer: a row tied with the boundary sits on its resolved side. + A boundary proves the object was gone at that instant, and a block that + cannot be ordered after the destruction must not seed a new "current" + streak — that would mint a survivor exempt from age-out forever for an + object that no longer exists. + """ + return sa.or_(boundary.c.boundary.is_(None), row.c.created_on > boundary.c.boundary) + + +def _repeats_an_earlier_block( + table: sa.Table, boundary: sa.Subquery +) -> sa.ColumnElement[bool]: + """Whether a same-reason current-streak block precedes this row with no + change of reason in between. + + This is the audit writer's suppression rule + (:func:`audit.finalize_retention_blocked`) applied retroactively: a + block repeating the reason of the block just before it adds nothing, + while the first block after a reason change is the only durable record + of the new cause and is retained. The comparison is NULL-safe, so + consecutive reason-less (pre-feature) rows count as one run, deduped to + the run's earliest, and the first coded block ends that run. Note this + is *stricter* than the writer for reason-less rows: the writer never + suppresses a reason-less block (``_suppress_redundant_block`` bails on a + missing code), so here the pruner additionally collapses legacy + pre-feature duplicates — the streak's earliest "blocked since" row is + still always kept. Tied same-reason rows are not "earlier" than each + other, so all of them are kept. + + A ``force`` row is exempt: it is never reported as a repeat (see the + return), so it is kept out of the duplicate category and marked a + survivor while its streak is current. It is also exempt from operational + age-out (see :func:`_operational_candidates`), so an operator force-purge + block is retained permanently — never pruned by either category. + """ + earlier: sa.FromClause = table.alias("earlier_block") + between: sa.FromClause = table.alias("reason_change") + reason_changed_between: sa.ColumnElement[bool] = sa.exists( + sa.select(sa.literal(1)) + .select_from(between) + .where( + sa.and_( + between.c.status == STATUS_BLOCKED, + between.c.entity_type == table.c.entity_type, + between.c.entity_uuid == table.c.entity_uuid, + # Inclusive bounds: a differing-reason block sharing an + # exact timestamp with either endpoint still breaks the run, + # so a reason-transition row tied with a neighbour is + # preserved as a run head rather than pruned as a repeat + # (the same preserving-side tie rule the pending and evidence + # guards use). Inclusive bounds only ever add boundaries — + # i.e. only ever preserve more, never delete more. + between.c.created_on >= earlier.c.created_on, + between.c.created_on <= table.c.created_on, + between.c.reason.is_distinct_from(table.c.reason), + ) + ) + .correlate(table, earlier) + ) + repeats: sa.ColumnElement[bool] = sa.exists( + sa.select(sa.literal(1)) + .select_from(earlier) + .where( + sa.and_( + earlier.c.status == STATUS_BLOCKED, + earlier.c.entity_type == table.c.entity_type, + earlier.c.entity_uuid == table.c.entity_uuid, + _in_current_streak(earlier, boundary), + earlier.c.created_on < table.c.created_on, + earlier.c.reason.is_not_distinct_from(table.c.reason), + sa.not_(reason_changed_between), + ) + ) + .correlate(table, boundary) + ) + # A ``force`` attempt is an operator action the writer never suppresses: + # audit.py's ``_suppress_redundant_block`` only collapses consecutive + # scheduled same-reason blocks, so the pruner does not collapse a force row + # either — it is never reported as a repeat. This keeps it out of the + # duplicate category and (via ``sa.not_`` in the operational category) marks + # it a survivor while its streak is current. It is also exempt from + # operational age-out (``_operational_candidates`` excludes force blocks), so + # a force block is retained permanently — full immortality for operator + # force-purge blocks. A force row may still be the *earlier* anchor a later + # scheduled repeat collapses into — only the force row itself is protected + # from duplicate removal. + return sa.and_(table.c.trigger != TRIGGER_FORCE, repeats) + + +def _preceded_by_unresolved_attempt(table: sa.Table) -> sa.ColumnElement[bool]: + """Whether an unresolved (``pending``) attempt precedes this row. + + A ``pending`` row is the only thing that can insert a streak boundary + into *history*: every other write lands at ``now``, newer than every + existing row, whereas reconciliation and the purge path finalize a + pending row **in place**, keeping its original ``created_on``. So a + pending row sitting between two blocked rows is a boundary that may + appear at any moment, and the blocked row after it would become the new + streak's survivor — the very row pruning must never delete. + + A tied timestamp counts as preceding. Which of the two writes landed + first is unknowable from the row, the block's classification (current + duplicate, or resolved-streak row on an age window) changes with the + attempt's outcome, and deferring it until then costs nothing. + """ + pending: sa.FromClause = table.alias("unresolved_attempt") + return sa.exists( + sa.select(sa.literal(1)) + .select_from(pending) + .where( + sa.and_( + pending.c.status == STATUS_PENDING, + pending.c.entity_type == table.c.entity_type, + pending.c.entity_uuid == table.c.entity_uuid, + pending.c.created_on <= table.c.created_on, + ) + ) + ) + + +def _duplicate_candidates(now: datetime, limit: int) -> sa.sql.Select: + """Select current-streak blocked rows that repeat the block before them. + + Age-independent by design (FR-003): a repeat is prunable the moment the + streak holds an earlier same-reason block, regardless of the retention + window. Trigger does not gate streak membership — ``scheduled`` and + ``force`` blocked rows share streaks — but a ``force`` row is itself never + collapsed as a repeat (see :func:`_repeats_an_earlier_block`). Reason *is* + a discriminator: the first block after a reason change survives alongside + the streak's earliest row. + + Rows preceded by an unresolved attempt are skipped: their classification + is not stable, because that attempt can finalize into a boundary and + promote them to survivor between selection and deletion. Such rows are + collected once the attempt resolves. Note that pruning does not depend on + that happening promptly — reconciliation runs from the purge task, which + a deployment may have disabled or left in dry-run — so a long-lived + pending row defers its successors indefinitely rather than risking them. + """ + table: sa.Table = PurgeAuditLog.__table__ + boundary: sa.Subquery = _streak_boundary_subquery(now) + return ( + sa.select(table.c.id) + .select_from(_with_boundary(table, boundary)) + .where(table.c.status == STATUS_BLOCKED) + .where(table.c.entity_uuid.is_not(None)) + .where(table.c.created_on <= now) + .where(_in_current_streak(table, boundary)) + .where(_repeats_an_earlier_block(table, boundary)) + .where(sa.not_(_preceded_by_unresolved_attempt(table))) + .order_by(table.c.created_on) + .limit(limit) Review Comment: The ID cap bounds the outer re-check, but each row still evaluates the history-sensitive `_repeats_an_earlier_block()` predicates while the coordination lock is held, and `reason` is not covered by either pruning index; a long multi-reason history can therefore keep `write_ahead()` blocked while those probes walk the entity's history. Could we add the promised representative query-plan/runtime evidence or make the locked validation independent of per-entity history? ########## UPDATING.md: ########## @@ -243,6 +243,41 @@ unknown impact as zero. Chart and dashboard purge endpoints are unchanged. - The dashboard datasource-based visibility fallback now fails closed: a dashboard whose member charts’ datasources cannot be resolved (deleted datasource rows, missing `datasource_id`, or unsupported datasource types) is no longer accessible to users without explicit editor/viewer rights, and a dashboard composed of semantic-view charts now requires `datasource_access` on (at least one of) its semantic views or their parent semantic layer — previously any authenticated user could open such a dashboard’s shell. Because the fallback now considers every member chart rather than only table-backed ones, a user holding `datasource_access` on any single member datasource — including a semantic view or its parent layer — can open a mixed dashboard that previously denied them. Dashboards with no charts remain accessible, and dashboards with explicit viewers are unaffected. Conversely, holders of `all_datasource_access` now see every published no-viewer dashboard in the dashboard l ist — including chart-less ones previously hidden by the inner joins — matching what the object-level gate already allowed them to open. - Version restore (`POST /api/v1/{chart,dashboard,dataset}/<uuid>/versions/<version_uuid>/restore`) now refuses an **externally managed** entity (`is_managed_externally = True`) with HTTP 403, enforcing server-side what the docs already promised. Previously the refusal existed only in the browser, so an otherwise-authorized editor could restore such an entity by calling the endpoint directly and have the restore overwritten on the next external sync. Soft-delete recovery is deliberately unaffected — it changes visibility, not content. +- The purge audit log can now be pruned automatically. The new + `deletion_retention.prune_purge_audit` Celery beat task (daily, 03:30, in the + default `CeleryConfig.beat_schedule`) removes duplicate `blocked` records + within an entity's current blockage streak (the earliest — "blocked since" — + record and the first record after each change of block `reason` always + survive, mirroring the audit writer's own suppression rule) and ages out + operational records (`blocked` from + resolved streaks, `failed`) older than + `PURGE_AUDIT_OPERATIONAL_RETENTION_DAYS` (default 90). A streak is ended + only by proof the object is gone (`confirmed`/`target_absent`); a `failed` + attempt does not reset the "blocked since" record. Completed-destruction + evidence (`confirmed`, `target_absent`) is **never touched** unless the + separate `PURGE_AUDIT_EVIDENCE_RETENTION_DAYS` opt-in is explicitly set, + which is the operator's assertion that an approved compliance policy + permits expiring destruction evidence. Automatic deletion is disabled by + default; set `PURGE_AUDIT_PRUNING_ENABLED = True` after reviewing these Review Comment: Enabling this in the same rolling deploy can violate the survivor invariant: workers on the base code stamp and commit `write_ahead()` rows without the coordination lock, so one can publish an earlier pending row after the new pruner's locked re-check/delete and later turn that deleted block into the post-boundary survivor. Could the rollout require every audit-writing worker to be on the coordinated protocol before this switch is enabled? ########## UPDATING.md: ########## @@ -945,7 +980,7 @@ With the flag on, delete confirmations across the chart/dashboard/dataset list p This also resolves the limitation noted under *Soft delete and restore for datasets*: a database blocked by soft-deleted datasets can now be freed by purging those datasets (per-entity endpoint, retention task, or `force-purge` CLI) instead of hard-deleting `tables` rows out-of-band. -The `purge_audit_log` table is **never pruned by design** — the audit must survive the entities it names; operators who need to age it out should prune manually. +Automatic pruning of the `purge_audit_log` table is available but **off by default**: set `PURGE_AUDIT_PRUNING_ENABLED = True` to enable the `deletion_retention.prune_purge_audit` Celery beat task (daily, 03:30) so the table no longer grows unbounded and does not need manual pruning. Left at its default (`PURGE_AUDIT_PRUNING_ENABLED = False`) the table is never pruned and grows indefinitely — enabling it is an explicit operator choice. The policy is written to preserve the audit's meaning rather than trade it away: within an entity's current blockage streak the earliest — "blocked since" — record always survives (only redundant duplicate `blocked` records are collapsed), and completed-destruction evidence (`confirmed`, `target_absent`) is **never** removed unless the separate `PURGE_AUDIT_EVIDENCE_RETENTION_DAYS` opt-in is explicitly set. What ages out is operational noise — `blocked` records from already-resolved streaks and `failed` records — once older than `PURGE_AUDIT _OPERATIONAL_RETENTION_DAYS` (default 90). See the release-note entry above for the beat-schedule and `CELERY_CONFIG` details. Review Comment: Enabling pruning still cannot keep this table bounded: every force-triggered `blocked` record is exempt from both duplicate removal and operational expiry, so repeated `force-purge` attempts against a persistently blocked entity grow it forever despite this sentence. Could this document the permanent force-block exception instead of promising that enabling the task makes growth bounded? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
