mikebridge commented on code in PR #43490: URL: https://github.com/apache/superset/pull/43490#discussion_r3931387146
########## superset/commands/deletion_retention/prune_audit.py: ########## @@ -0,0 +1,689 @@ +# 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, + 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. + """ + 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) + ) + return 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) + ) + + +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 is not a discriminator — ``scheduled`` and ``force`` + blocked rows share streaks. Reason *is*: the first block after a reason + change survives alongside the streak's earliest row (see + :func:`_repeats_an_earlier_block`). + + 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) Review Comment: You're right — this one is **real**. Built your scenario (`blocked(scheduled, R) → blocked(force, R)`, same entity) as an integration test against head `e2dddd9da7`: the force block **was pruned** as a duplicate of the earlier scheduled block. `_duplicate_candidates` treated trigger as a non-discriminator, so it collapsed the force row even though the writer retains force outcomes independently. **Fixed in `ac9bb3bf34`:** `_repeats_an_earlier_block` now excludes force-trigger rows — the pruner mirrors the writer, whose `_suppress_redundant_block` only collapses consecutive *scheduled* same-reason blocks. A force row is never reported as a repeat, so it's kept out of the duplicate category and marked a survivor while its streak is current. `test_force_block_after_scheduled_same_reason_survives` now passes (it failed before the fix). A force row can still be the *earlier* anchor a later scheduled repeat collapses into — `test_force_anchor_still_collapses_a_later_scheduled_repeat` proves the exemption is surgical (only the force row itself is protected). 34 unit + 117 integration tests green. **One scope boundary I want your explicit call on** (a 4-lens pass surfaced it, not in your original scenario): the exemption protects a force block from duplicate collapse and from current-streak age-out, but **not** once its streak resolves. A force block sitting behind a `confirmed`/`target_absent` boundary still ages on the operational window like any resolved-streak blocked row (probed: it is deleted). That's consistent with the module's taxonomy — blocked rows are operational and decay; only `confirmed`/`target_absent` are durable evidence, and the boundary records the destruction — so I did **not** extend the exemption to operational aging (that would contradict the documented taxonomy and go beyond this finding). I scoped the docstring/comment accordingly and added `test_resolved_streak_force_block_ages_on_operational_window` to pin it. If you'd rather force records be fully immortal (never auto-aged even in a resolved streak), say so and I'll extend the exemption to t he operational category — it's a small, safe change, just a policy call I want you to make on an irreversible feature. -- 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]
