This is an automated email from the ASF dual-hosted git repository.
aminghadersohi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new 38ad1150ecf feat(retention): cap soft-delete purge and version-history
prune per run (#45029)
38ad1150ecf is described below
commit 38ad1150ecf03dab064ccbd4f5601b4de84102d7
Author: Mike Bridge <[email protected]>
AuthorDate: Sat Oct 10 16:26:50 2026 -0600
feat(retention): cap soft-delete purge and version-history prune per run
(#45029)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
UPDATING.md | 4 +
docs/docs/using-superset/recently-archived.mdx | 37 +
docs/docs/using-superset/version-history.mdx | 33 +-
superset/config.py | 8 +
superset/tasks/deletion_retention.py | 397 +++++++---
superset/tasks/retention_cap.py | 32 +
superset/tasks/version_history_retention.py | 282 ++++++--
.../versioning/retention_prune_tests.py | 65 ++
.../deletion_retention/test_purge_chart_detach.py | 18 +-
tests/unit_tests/tasks/test_deletion_retention.py | 796 ++++++++++++++++++++-
.../tasks/test_version_history_retention.py | 659 ++++++++++++++++-
11 files changed, 2142 insertions(+), 189 deletions(-)
diff --git a/UPDATING.md b/UPDATING.md
index fd50555be96..40ef0301342 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -1841,6 +1841,8 @@ Authorization reuses the resource's `can_read` permission
and per-object `raise_
Entity version history (the `version_transaction` / `*_version` shadow tables
that back version capture) is aged out by a nightly Celery beat task,
`version_history.prune_old_versions`
(`superset.tasks.version_history_retention`).
+**Scheduled-run cap (behavior change):**
`VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN` defaults to 1000 whole
prunable transactions per invocation; associated shadow and change rows remain
atomic. `0` or `None` restores unlimited runs. Invalid values skip the task
before deletion. Capped results report `cap_reached`, `remaining_eligible`, and
`remaining_count_complete`; the remainder is a lower bound from one candidate
window after the live scan's stopping point when the probe does no [...]
+
| Key | Default | Purpose |
|---|---|---|
| `VERSION_HISTORY_RETENTION_DAYS` | `30` | Version rows whose owning
`version_transaction.issued_at` is older than this many days are pruned. Each
entity's live row (`end_transaction_id IS NULL`) is always preserved, as are
the live rows of its children and associations; closed historical rows
(including the baseline) age out. `0` disables pruning; `-1` makes historical
rows eligible on the next scheduled run. Other negative values are invalid and
skip pruning. |
@@ -1864,6 +1866,8 @@ Soft-deleted dashboards, charts, and datasets are now
permanently removed after
Purging is **live by default** (`SOFT_DELETE_PURGE_DRY_RUN=False`), so the
retention promise above is real on a stock deployment. Set it to `True` to have
the task log `would_purge` counts and delete nothing — the lever is retained,
so an operator can return to dry-run at any time. 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 a [...]
+**Scheduled-run cap (behavior change):** `SOFT_DELETE_PURGE_MAX_PER_RUN`
defaults to 1000 successful root-entity purges across charts, dashboards, and
datasets per invocation. One root plus its cascade counts as one; physical row
counts are separate. The cap bounds committed deletions, not candidate
evaluations: blocked roots are re-evaluated on every run, so a blocked-heavy
backlog can still require substantial work. Set `0` or `None` for unlimited
scheduled purging. Invalid values skip [...]
+
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 scheduled
evaluation writes a provisional, content-free record to the new
`purge_audit_log` table before the cascade s [...]
diff --git a/docs/docs/using-superset/recently-archived.mdx
b/docs/docs/using-superset/recently-archived.mdx
index da8c5d8c687..b64e94f9a90 100644
--- a/docs/docs/using-superset/recently-archived.mdx
+++ b/docs/docs/using-superset/recently-archived.mdx
@@ -292,6 +292,43 @@ destructive when it runs. Set it to `True` in
`superset_config.py` and restart
the relevant processes to log what the task *would* purge without deleting
anything. Validate the eligible backlog before enabling live purging or
shortening a window; the first live run can process accumulated old archives.
+`SOFT_DELETE_PURGE_MAX_PER_RUN` defaults to `1000` root charts, dashboards, or
+datasets per scheduled run. One root's complete cascade counts as one and is
+never split to fit the cap. Set `0` or `None` in `superset_config.py` to remove
+the cap. It bounds committed deletions, not candidate evaluations: blocked
roots
+are re-evaluated every run, so a blocked-heavy backlog can still require
+substantial work. A negative, boolean, string, or fractional value stops the
+scheduled task without purging. With a positive cap configured, a live run
+reports `cap_reached` and `remaining_eligible` for aged roots remaining across
+all supported models. In unlimited mode, the cap and remainder fields and
+their gauges are not emitted. The capped remainder includes
+roots that a reference blocker may prevent from purging. A remainder-count
failure,
+a model scan failure, or an uncertain commit sets `remaining_eligible` to
`None`
+and `remaining_count_complete` to `False`; this can happen even when no purge
has
+been confirmed. Confirmed purge totals are retained. Inspect `scan_failures`,
+`commit_uncertain`, and the `deletion_retention.remainder_count_failed` counter
+and task logs to distinguish those cases. Dry-run scans the full
+eligible backlog regardless of the cap, writes nothing, and reports
+`eligible_backlog` plus `estimated_capped_runs`; the estimate assumes all
+eligible roots can be purged. If a model scan fails, `would_purge` contains
only
+the counts collected before that failure; `backlog_count_complete` is `False`
+and both backlog estimates are `None`. If the database does not acknowledge a
+root purge commit, the task stops without processing another root and reports
+`commit_uncertain`. The possible deletion reserves a cap slot, its audit row
+remains pending for reconciliation, and the remainder count is incomplete.
+A commit the database rejects on a live connection is instead recorded as a
+failed attempt for that root, counted in `cascade_failures`, and the task
+continues with other eligible roots under the same cap.
+The operator's `force-purge` CLI remains uncapped.
+
+With the default daily schedule and cap, scheduled capacity is 1,000 root
+entities per day, not 1,000 physical rows. A backlog drains only when
successful
+purges exceed newly eligible roots: at that throughput, 1,500 newly eligible
+roots per day add 500 to the backlog per day. Size
`SOFT_DELETE_PURGE_MAX_PER_RUN` in
+`superset_config.py`, or increase the task's beat frequency, above eligible
inflow
+with catch-up headroom. Repeated `cap_reached=True` results warrant checking
the
+backlog trend and capacity. Blocked roots and failures can reduce actual
throughput;
+increasing the cap alone does not remove reference blockers.
### Running retention tasks
diff --git a/docs/docs/using-superset/version-history.mdx
b/docs/docs/using-superset/version-history.mdx
index 1986cf30a07..3f7cc8c63e0 100644
--- a/docs/docs/using-superset/version-history.mdx
+++ b/docs/docs/using-superset/version-history.mdx
@@ -96,6 +96,33 @@ History retention is configured independently with
`VERSION_HISTORY_RETENTION_DA
in `superset_config.py` or the environment variable of the same name. Its
default
is 30 days; set it to `0` to disable pruning. The scheduled retention task
preserves
live versions and prunes eligible closed history older than the configured
window.
+`VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN` defaults to `1000` whole
+transactions per scheduled run. The transaction's shadow rows and change
records
+are removed together; the cap counts transactions, not physical rows. Set the
+cap to `0` or `None` in `superset_config.py` for an unlimited run. Negative,
+boolean, string, and fractional caps are invalid and stop the task without
+deleting history. A capped run reports `cap_reached`, `remaining_eligible`, and
+`remaining_count_complete`. The remainder is exact when the bounded probe
+covers the backlog; otherwise it is a lower bound from its first 1,000 aged
+transaction candidates after the live scan's stopping point. If that probe
+fails after a successful prune, the remainder is `None` without hiding the
+committed deletion totals. A backlog drains only when successful pruning
exceeds
+the rate at which transactions become eligible. With the default daily schedule
+and cap, scheduled capacity is 1,000 transactions per day. At that throughput,
+an eligible inflow of 1,500 per day grows the backlog by 500 per day. Size the
cap in
+`superset_config.py`, or increase the task's beat frequency, to exceed eligible
+inflow and leave capacity for catch-up. Repeated `cap_reached=True` results
warrant
+checking the backlog trend and available capacity; read them alongside
remainder
+completeness and task errors. Live-history exclusions can reduce actual
throughput.
+If a database commit's outcome is uncertain, the task stops rather than
+retrying that window, so it cannot spend another cap allowance in the same
+run. The task reports an error because it cannot assert how many transactions
+were committed. A definitive `OperationalError`, including one rejected at
+commit with a valid connection, may retry up to three attempts.
+Monitor the
+`superset.versioning.retention.remaining_count_complete` gauge alongside the
+`remaining_eligible_at_least` gauge so a lower-bound zero is not read as an
+empty backlog.
The released `SUPERSET_VERSION_HISTORY_RETENTION_DAYS` key remains a deprecated
fallback: its positive window is preserved and any nonpositive value still
disables pruning. Migrate a legacy `-1` or other negative disable to the new
@@ -121,7 +148,11 @@ one of those transactions: closed rows can be removed when
their closing
transaction expires. Retention can therefore remove history used by previews
and restores; version history is not a substitute for backups.
-There is no dry-run switch for this task. Neither `VERSION_HISTORY` nor
+Set `VERSION_HISTORY_PRUNE_DRY_RUN = True` in `superset_config.py` to scan the
+entire eligible backlog without deleting anything. The result includes
+`eligible_backlog` and `estimated_capped_runs`; a live-anchor transaction is
+excluded from both. The dry-run setting must be a boolean; an invalid value
+skips the scheduled task without deleting history. Neither `VERSION_HISTORY`
nor
`ENABLE_VERSIONING_CAPTURE` disables pruning of existing history. To preserve
history while the feature is off, set the retention value to `0` and leave the
task's beat schedule entry in place: removing the entry logs a missing-schedule
diff --git a/superset/config.py b/superset/config.py
index f311d533331..176ccf6c772 100644
--- a/superset/config.py
+++ b/superset/config.py
@@ -1080,6 +1080,9 @@ SOFT_DELETE_RETENTION_DAYS_FUNC: Callable[[], int] | None
= None
# built-in chart, dashboard and dataset roots are never affected.
PURGE_POLICIES_FUNC: Callable[[], Any] | None = None
SOFT_DELETE_PURGE_DRY_RUN: bool = False
+# Maximum committed root purges per scheduled run; 0 or None is unlimited.
+# Set an int in superset_config.py; this key has no automatic environment
parsing.
+SOFT_DELETE_PURGE_MAX_PER_RUN: int | None = 1000
# Retention policy for the purge audit log itself (the durable evidence the
# purge task writes). Pruning is deletion-only and scheduled
@@ -1908,6 +1911,11 @@ def _normalize_version_history_retention_days(value:
object, *, legacy: bool) ->
VERSION_HISTORY_RETENTION_DAYS: int = _parse_version_history_retention_days()
+# Maximum committed transactions pruned per scheduled run; 0 or None is
unlimited.
+# Set an int in superset_config.py; this key has no automatic environment
parsing.
+VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN: int | None = 1000
+# Set a bool in superset_config.py; this key has no automatic environment
parsing.
+VERSION_HISTORY_PRUNE_DRY_RUN: bool = False
_version_history_retention_seed: int = VERSION_HISTORY_RETENTION_DAYS
# Sentinel for "key not configured at all", distinct from any configured value
diff --git a/superset/tasks/deletion_retention.py
b/superset/tasks/deletion_retention.py
index 6db107cec7d..5ff3e879855 100644
--- a/superset/tasks/deletion_retention.py
+++ b/superset/tasks/deletion_retention.py
@@ -21,22 +21,23 @@ that ages out version rows while keeping the live entity,
this removes
entities that are already soft-deleted. For each
``SoftDeleteMixin`` model with a purge policy 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.
+cascade per entity, in bounded id-ordered batches. Safe to repeat: each capped
+run purges at most the configured number of root entities. Successive runs
+converge by draining the purgeable backlog; blocked roots can remain, and
+newly eligible rows can add to that backlog.
"""
from __future__ import annotations
import logging
from collections.abc import Iterator
-from dataclasses import dataclass, field
from datetime import datetime, timedelta
-from typing import Any, cast
+from typing import Any, cast, NamedTuple
from uuid import UUID
import sqlalchemy as sa
from flask import current_app
+from sqlalchemy.exc import DBAPIError
from superset import db
from superset.commands.deletion_retention import audit, prune_audit
@@ -57,6 +58,7 @@ from superset.models.helpers import (
skip_visibility_filter,
SoftDeleteMixin,
)
+from superset.tasks.retention_cap import validate_retention_cap
logger: logging.Logger = logging.getLogger(__name__)
@@ -71,6 +73,15 @@ def _soft_delete_models() -> list[type[SoftDeleteMixin]]:
return list(SoftDeleteMixin._registered_subclasses) # noqa: SLF001
+def _ordered_purge_models(cutoff: datetime) -> list[type[SoftDeleteMixin]]:
+ """Rotate daily priority so one busy model cannot starve later models."""
+ models: list[type[SoftDeleteMixin]] = _soft_delete_models()
+ if not models:
+ return models
+ start: int = cutoff.toordinal() % len(models)
+ return models[start:] + models[:start]
+
+
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__)
@@ -133,35 +144,73 @@ def _report_model_counts(outcome: str, counts: dict[str,
int]) -> None:
)
-@dataclass
-class _PassTotals:
- """What one pass over the soft-delete roots produced."""
+def _count_eligible(model: type[SoftDeleteMixin], cutoff: datetime) -> int:
+ """Count aged roots, including roots that a later cascade may block."""
+ table: sa.Table = _model_table(model)
+ with skip_visibility_filter(db.session, model):
+ return int(
+ db.session.scalar(
+ sa.select(sa.func.count())
+ .select_from(table)
+ .where(table.c.deleted_at.is_not(None), table.c.deleted_at <
cutoff)
+ )
+ or 0
+ )
- purged: dict[str, int] = field(default_factory=dict)
- would_purge: dict[str, int] = field(default_factory=dict)
- unsupported: dict[str, int] = field(default_factory=dict)
- cascade_failures: int = 0
- blocked: int = 0
+
+class _PurgeScan(NamedTuple):
+ """Counts and supported models collected across one purge invocation."""
+
+ purged: dict[str, int]
+ would_purge: dict[str, int]
+ unsupported_models: dict[str, int]
+ failures: int
+ blocked: int
+ remaining_budget: int | None
+ supported_models: list[type[SoftDeleteMixin]]
scan_failures: int = 0
- #: Roots this pass actually reached, so a run where every one of them
- #: failed can be told apart from a run where one did.
attempted: int = 0
+ commit_uncertain: bool = False
+
+
+class _PurgeModelResult(NamedTuple):
+ """Named per-root counts, including an indeterminate commit attempt."""
+
+ purged: int
+ would_purge: int
+ failures: int
+ blocked: int
+ scan_failures: int
+ commit_uncertain: bool = False
-def _purge_roots(cutoff: datetime, dry_run: bool) -> _PassTotals:
- """Process each registered root, isolating one root's failure from the
rest."""
- totals = _PassTotals()
- for model in _soft_delete_models():
- totals.attempted += 1
- # The table name is itself read off the model, so resolving it belongs
- # inside the guard: a root that cannot supply one would otherwise end
- # the pass here, above the isolation meant to contain it. The class
- # name stands in until it is known, for the failure log.
- entity_type = model.__name__
+class _PurgeCommitUncertainError(RuntimeError):
+ """The database did not acknowledge whether a root purge committed."""
+
+
+def _scan_purge_models(
+ cutoff: datetime, dry_run: bool, max_per_run: int | None
+) -> _PurgeScan:
+ """Apply one shared root budget across every supported model."""
+ purged: dict[str, int] = {}
+ would_purge: dict[str, int] = {}
+ unsupported_models: dict[str, int] = {}
+ failures: int = 0
+ blocked: int = 0
+ scan_failures: int = 0
+ attempted: int = 0
+ commit_uncertain: bool = False
+ remaining_budget: int | None = max_per_run
+ supported_models: list[type[SoftDeleteMixin]] = []
+
+ for model in _ordered_purge_models(cutoff):
+ attempted += 1
+ entity_type: str = model.__name__
try:
entity_type = _model_table_name(model)
if model not in purge_policy_registry():
- totals.unsupported[entity_type] = 1
+ attempted -= 1
+ unsupported_models[entity_type] = 1
logger.warning(
"deletion_retention: skipping %s: no purge policy",
entity_type
)
@@ -169,16 +218,19 @@ def _purge_roots(cutoff: datetime, dry_run: bool) ->
_PassTotals:
f"{_METRIC_PREFIX}.unsupported_models.{entity_type}"
)
continue
- purged_n, would_n, failed_n, blocked_n, scan_failed_n =
_purge_model(
- model, cutoff, dry_run
+ supported_models.append(model)
+ if remaining_budget == 0 and not dry_run:
+ attempted -= 1
+ continue
+ counts: _PurgeModelResult = _purge_model(
+ model,
+ cutoff,
+ dry_run,
+ max_per_run=None if dry_run else remaining_budget,
)
except Exception: # pylint: disable=broad-except
- # One root must not cost the others their run. _purge_model keeps
- # its own counts when a scan fails part-way; this guards what is
- # left -- resolving the registry, and anything else before the
- # first page.
db.session.rollback() # pylint: disable=consider-using-transaction
- totals.scan_failures += 1
+ scan_failures += 1
stats_logger_manager.instance.incr(
f"{_METRIC_PREFIX}.scan_failures.{entity_type}"
)
@@ -186,18 +238,87 @@ def _purge_roots(cutoff: datetime, dry_run: bool) ->
_PassTotals:
"deletion_retention: %s could not be processed", entity_type
)
continue
- if would_n:
- totals.would_purge[entity_type] = would_n
- if purged_n:
- totals.purged[entity_type] = purged_n
- totals.cascade_failures += failed_n
- totals.blocked += blocked_n
- totals.scan_failures += scan_failed_n
- if scan_failed_n:
+ if remaining_budget is not None and not dry_run:
+ # Reserve the possible commit even though it is not confirmed.
+ remaining_budget -= counts.purged + int(counts.commit_uncertain)
+ if counts.would_purge:
+ would_purge[entity_type] = counts.would_purge
+ if counts.purged:
+ purged[entity_type] = counts.purged
+ failures += counts.failures
+ blocked += counts.blocked
+ scan_failures += counts.scan_failures
+ if counts.scan_failures:
stats_logger_manager.instance.incr(
f"{_METRIC_PREFIX}.scan_failures.{entity_type}"
)
- return totals
+ if counts.commit_uncertain:
+ commit_uncertain = True
+ break
+
+ return _PurgeScan(
+ purged=purged,
+ would_purge=would_purge,
+ unsupported_models=unsupported_models,
+ failures=failures,
+ blocked=blocked,
+ remaining_budget=remaining_budget,
+ supported_models=supported_models,
+ scan_failures=scan_failures,
+ attempted=attempted,
+ commit_uncertain=commit_uncertain,
+ )
+
+
+def _add_purge_cap_stats(
+ stats: dict[str, Any],
+ cutoff: datetime,
+ scan: _PurgeScan,
+ max_per_run: int | None,
+) -> None:
+ """Count remaining roots without masking a committed purge on failure."""
+ if max_per_run is None:
+ return
+ remaining_eligible: int | None
+ count_complete: bool = scan.scan_failures == 0 and not
scan.commit_uncertain
+ if count_complete:
+ try:
+ remaining_eligible = sum(
+ _count_eligible(model, cutoff) for model in
scan.supported_models
+ )
+ except Exception: # pylint: disable=broad-except
+ logger.warning(
+ "deletion_retention: remainder count failed after purge",
+ exc_info=True,
+ )
+ try:
+ db.session.rollback() # pylint:
disable=consider-using-transaction
+ except Exception: # pylint: disable=broad-except
+ logger.exception("deletion_retention: remainder count rollback
failed")
+ stats_logger_manager.instance.incr(
+ f"{_METRIC_PREFIX}.remainder_count_failed"
+ )
+ remaining_eligible = None
+ count_complete = False
+ else:
+ remaining_eligible = None
+ cap_reached: bool = scan.remaining_budget == 0
+ stats.update(
+ max_per_run=max_per_run,
+ cap_reached=cap_reached,
+ remaining_eligible=remaining_eligible,
+ remaining_count_complete=count_complete,
+ )
+ if remaining_eligible is not None:
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.remaining_eligible", remaining_eligible
+ )
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.remaining_count_complete", int(count_complete)
+ )
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.cap_reached", int(cap_reached)
+ )
def _report_total_outage(every_root_failed: bool) -> None:
@@ -217,8 +338,11 @@ def _report_total_outage(every_root_failed: bool) -> None:
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.failed")
-def _purge_impl(window_days: int, dry_run: bool) -> dict[str, Any]:
+def _purge_impl(
+ window_days: int, dry_run: bool, max_per_run: int | None = None
+) -> dict[str, Any]:
"""Run one purge pass across all soft-delete models."""
+ max_per_run = validate_retention_cap(max_per_run,
"SOFT_DELETE_PURGE_MAX_PER_RUN")
if window_days == 0 or window_days < -1:
logger.info("deletion_retention: window is disabled or invalid;
skipping")
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped")
@@ -236,62 +360,76 @@ def _purge_impl(window_days: int, dry_run: bool) ->
dict[str, Any]:
else datetime.now() - timedelta(days=window_days)
)
_reconcile_unless_dry_run(dry_run)
- totals: _PassTotals = _purge_roots(cutoff, dry_run)
- purged = totals.purged
- would_purge = totals.would_purge
- unsupported_models = totals.unsupported
- failures = totals.cascade_failures
- blocked = totals.blocked
- scan_failures = totals.scan_failures
- # A root with no policy was never scanned, so it is not a failure and
- # does not count toward "every root failed" -- which is what tells a
- # metadata-database outage apart from one root misbehaving.
- scannable: int = totals.attempted - len(unsupported_models)
- every_root_failed: bool = scannable > 0 and scan_failures == scannable
+ scan: _PurgeScan = _scan_purge_models(cutoff, dry_run, max_per_run)
+ every_root_failed: bool = (
+ scan.attempted > 0 and scan.scan_failures == scan.attempted
+ )
if dry_run:
- _report_model_counts("would_purge", would_purge)
- logger.info("deletion_retention: DRY RUN would_purge=%s", would_purge)
- if scan_failures:
- stats_logger_manager.instance.gauge(
- f"{_METRIC_PREFIX}.scan_failures", scan_failures
+ _report_model_counts("would_purge", scan.would_purge)
+ logger.info("deletion_retention: DRY RUN would_purge=%s",
scan.would_purge)
+ backlog: int | None = (
+ sum(scan.would_purge.values()) if scan.scan_failures == 0 else None
+ )
+ estimated_runs: int | None = None
+ if backlog is not None:
+ estimated_runs = (
+ (backlog + max_per_run - 1) // max_per_run
+ if max_per_run is not None
+ else int(backlog > 0)
)
- _report_total_outage(every_root_failed)
- return {
+ dry_stats: dict[str, Any] = {
"dry_run": 1,
- "would_purge": would_purge,
- "unsupported_models": unsupported_models,
- "scan_failures": scan_failures,
+ "would_purge": scan.would_purge,
+ "unsupported_models": scan.unsupported_models,
+ "scan_failures": scan.scan_failures,
+ "eligible_backlog": backlog,
+ "backlog_count_complete": backlog is not None,
+ "max_per_run": max_per_run,
+ "estimated_capped_runs": estimated_runs,
}
+ if scan.scan_failures:
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.scan_failures", scan.scan_failures
+ )
+ _report_total_outage(every_root_failed)
+ return dry_stats
- _report_model_counts("purged", purged)
- if failures:
+ _report_model_counts("purged", scan.purged)
+ if scan.failures:
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.cascade_failures")
- if blocked:
+ if scan.blocked:
stats_logger_manager.instance.gauge(
- f"{_METRIC_PREFIX}.blocked_by_reference", blocked
+ f"{_METRIC_PREFIX}.blocked_by_reference", scan.blocked
)
- if scan_failures:
+ if scan.scan_failures:
stats_logger_manager.instance.gauge(
- f"{_METRIC_PREFIX}.scan_failures", scan_failures
+ f"{_METRIC_PREFIX}.scan_failures", scan.scan_failures
)
+ if scan.commit_uncertain:
+
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.commit_uncertain")
+ stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.failed")
_report_total_outage(every_root_failed)
- stats = {
- "purged": purged,
- "cascade_failures": failures,
- "blocked_by_reference": blocked,
- "unsupported_models": unsupported_models,
- "scan_failures": scan_failures,
+ stats: dict[str, Any] = {
+ "purged": scan.purged,
+ "cascade_failures": scan.failures,
+ "blocked_by_reference": scan.blocked,
+ "unsupported_models": scan.unsupported_models,
+ "scan_failures": scan.scan_failures,
+ "commit_uncertain": scan.commit_uncertain,
}
+ _add_purge_cap_stats(stats, cutoff, scan, max_per_run)
logger.info("deletion_retention: %s", stats)
return stats
-def _purge_model(
- model: type[SoftDeleteMixin], cutoff: datetime, dry_run: bool
-) -> tuple[int, int, int, int, int]:
- """Process one model's eligible rows. Returns ``(purged, would_purge,
- failures, blocked, scan_failures)``. A single entity's blocked/failed
+def _purge_model( # noqa: C901
+ model: type[SoftDeleteMixin],
+ cutoff: datetime,
+ dry_run: bool,
+ max_per_run: int | None = None,
+) -> _PurgeModelResult:
+ """Process one model's eligible rows. A single entity's blocked/failed
cascade never aborts the batch.
A failure in the eligible-id scan itself -- a column it cannot read, a
@@ -302,18 +440,30 @@ def _purge_model(
"""
entity_type = _model_table_name(model)
purged = would = failures = blocked = scan_failures = 0
+ commit_uncertain: bool = False
try:
for id_batch in _iter_eligible_ids(model, cutoff, _BATCH):
if dry_run:
would += len(id_batch)
continue
for entity_id in id_batch:
+ if max_per_run is not None and purged >= max_per_run:
+ break
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 _PurgeCommitUncertainError:
+ commit_uncertain = True
+ logger.exception(
+ "deletion_retention: commit outcome unknown for %s
id=%s; "
+ "deferring remaining roots",
+ entity_type,
+ entity_id,
+ )
+ break
except Exception: # pylint: disable=broad-except
db.session.rollback() # pylint:
disable=consider-using-transaction
failures += 1
@@ -322,11 +472,21 @@ def _purge_model(
entity_type,
entity_id,
)
+ if commit_uncertain or (max_per_run is not None and purged >=
max_per_run):
+ break
except Exception: # pylint: disable=broad-except
- db.session.rollback() # pylint: disable=consider-using-transaction
scan_failures = 1
logger.exception("deletion_retention: scan failed for %s", entity_type)
- return purged, would, failures, blocked, scan_failures
+ try:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ except Exception: # pylint: disable=broad-except
+ # Earlier roots already committed; preserve their budget
accounting.
+ logger.exception(
+ "deletion_retention: scan rollback failed for %s", entity_type
+ )
+ return _PurgeModelResult(
+ purged, would, failures, blocked, scan_failures, commit_uncertain
+ )
def _finalize_blocked(record_id: UUID | None, blocker: BlockerReason) -> None:
@@ -342,6 +502,48 @@ def _finalize_blocked(record_id: UUID | None, blocker:
BlockerReason) -> None:
)
+def _confirm_committed_purge(record_id: UUID | None, result: CascadeResult) ->
None:
+ """Keep a committed root counted even if audit finalization fails."""
+ try:
+ audit.confirm(
+ record_id,
+ affected_referrers=result.dangling_chart_uuids,
+ removed_dashboard_slices=result.removed_dashboard_slices,
+ )
+ except Exception: # pylint: disable=broad-except
+ # The audit remains pending for its normal reconciliation path.
+ logger.exception(
+ "deletion_retention: audit finalization failed after committed
purge"
+ )
+
+
+def _commit_purge_root(record_id: UUID | None) -> None:
+ """Commit a purge, keeping uncertain outcomes pending for
reconciliation."""
+ try:
+ db.session.commit() # pylint: disable=consider-using-transaction
+ except Exception as exc:
+ try:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ except Exception: # pylint: disable=broad-except
+ logger.exception("deletion_retention: rollback after commit error
failed")
+ raise _PurgeCommitUncertainError(
+ "root purge rollback outcome unknown"
+ ) from exc
+ if isinstance(exc, DBAPIError) and not exc.connection_invalidated:
+ # The valid connection rejected the commit; the root can be
+ # recorded as failed without consuming the confirmed purge cap.
+ logger.warning(
+ "deletion_retention: definitive root purge commit failure
(%s): %s",
+ type(exc).__name__,
+ exc.orig,
+ )
+ audit.fail(record_id)
+ raise
+ # A lost acknowledgement can follow a successful commit. Keep the
+ # audit pending and defer remaining roots for reconciliation.
+ raise _PurgeCommitUncertainError("root purge commit outcome unknown")
from exc
+
+
def _purge_one(
model: type[SoftDeleteMixin], entity_id: int, cutoff: datetime
) -> CascadeResult | None:
@@ -403,19 +605,16 @@ def _purge_one(
# 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
+ # Commit/rollback are managed manually so audit.fail() can record a
+ # definitive pre-commit failure after the purge transaction resolves.
+ db.session.flush()
except Exception:
db.session.rollback() # pylint: disable=consider-using-transaction
audit.fail(record_id)
raise
+ _commit_purge_root(record_id)
if result.purged:
- audit.confirm(
- record_id,
- affected_referrers=result.dangling_chart_uuids,
- removed_dashboard_slices=result.removed_dashboard_slices,
- )
+ _confirm_committed_purge(record_id, result)
elif result.blocker is not None:
_finalize_blocked(record_id, result.blocker)
else:
@@ -498,7 +697,8 @@ def prune_purge_audit() -> dict[str, Any]:
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."""
+ not poison the schedule. The cap bounds committed root deletions, not
+ candidate evaluations; blocked roots are re-evaluated on every run."""
# 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).
@@ -508,7 +708,16 @@ def purge_soft_deleted() -> dict[str, Any]:
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)
+ max_per_run: int | None = validate_retention_cap(
+ current_app.config.get("SOFT_DELETE_PURGE_MAX_PER_RUN", 1000),
+ "SOFT_DELETE_PURGE_MAX_PER_RUN",
+ )
+ except ValueError:
+ logger.warning("deletion_retention: invalid purge cap; skipping")
+
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped_invalid_cap")
+ return {"skipped_invalid_cap": 1}
+ try:
+ return _purge_impl(window_days, dry_run, max_per_run=max_per_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")
diff --git a/superset/tasks/retention_cap.py b/superset/tasks/retention_cap.py
new file mode 100644
index 00000000000..dda3acc2f8d
--- /dev/null
+++ b/superset/tasks/retention_cap.py
@@ -0,0 +1,32 @@
+# 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.
+"""Validate per-invocation retention budgets before destructive work."""
+
+from __future__ import annotations
+
+
+def validate_retention_cap(value: object, key: str) -> int | None:
+ """Return a positive unit cap, or ``None`` for explicit unlimited mode.
+
+ Invalid configuration raises so scheduled cleanup fails closed. Booleans
+ and numeric strings are not valid integer budgets.
+ """
+ if value is None or (type(value) is int and value == 0):
+ return None
+ if type(value) is not int or value < 0:
+ raise ValueError(f"{key} must be a nonnegative integer or None")
+ return value
diff --git a/superset/tasks/version_history_retention.py
b/superset/tasks/version_history_retention.py
index a8d6cf44679..6b77a790abd 100644
--- a/superset/tasks/version_history_retention.py
+++ b/superset/tasks/version_history_retention.py
@@ -40,7 +40,9 @@ otherwise-expired history. Every other prunable transaction
is dropped, and
its ``version_changes`` rows cascade via the FK.
Registered via ``CeleryConfig.beat_schedule`` in ``superset/config.py``.
-Idempotent: a second run prunes nothing.
+Safe to repeat: each capped run prunes at most the configured number of
+transactions. Successive runs converge by draining the remaining prunable
+backlog; newly eligible transactions can add to that backlog.
"""
from __future__ import annotations
@@ -54,7 +56,7 @@ from typing import Any
import sqlalchemy as sa
from flask import current_app
-from sqlalchemy.exc import OperationalError
+from sqlalchemy.exc import DBAPIError, OperationalError
from superset.config import (
_MISSING_RETENTION,
@@ -62,6 +64,7 @@ from superset.config import (
_version_history_retention_seed,
)
from superset.extensions import celery_app, db, stats_logger_manager
+from superset.tasks.retention_cap import validate_retention_cap
from superset.utils.dates import naive_utcnow
logger: logging.Logger = logging.getLogger(__name__)
@@ -351,8 +354,56 @@ _RETRY_BACKOFF_FACTOR: int = 4
_METRIC_PREFIX: str = "superset.versioning.retention"
-def _run_prune_pass(
+def _live_bearing_tables(tables: ShadowTables) -> list[sa.Table]:
+ """Return every shadow table that can anchor a live transaction."""
+ live_bearing_tables: list[sa.Table] = [*tables.parent, *tables.child]
+ if tables.m2m is not None:
+ live_bearing_tables.append(tables.m2m)
+ return live_bearing_tables
+
+
+def _count_prunable(cutoff: datetime, tables: ShadowTables) -> int:
+ """Count the full eligible transaction backlog without writing rows."""
+ total: int = 0
+ after_id: int = 0
+ live_bearing_tables: list[sa.Table] = _live_bearing_tables(tables)
+ conn: sa.engine.Connection
+ while True:
+ with db.engine.connect() as conn:
+ window: _PruneWindow = _resolve_prune_window(
+ conn, cutoff, live_bearing_tables, after_id, _MAX_PRUNE_BATCH
+ )
+ total += len(window.prunable)
+ if window.candidate_count < _MAX_PRUNE_BATCH:
+ return total
+ after_id = window.max_candidate_id
+
+
+def _probe_prunable(
cutoff: datetime, tables: ShadowTables, after_id: int = 0
+) -> tuple[int, bool]:
+ """Count one candidate window after the live run's scan cursor."""
+ conn: sa.engine.Connection
+ with db.engine.connect() as conn:
+ window: _PruneWindow = _resolve_prune_window(
+ conn, cutoff, _live_bearing_tables(tables), after_id,
_MAX_PRUNE_BATCH
+ )
+ return len(window.prunable), window.candidate_count < _MAX_PRUNE_BATCH
+
+
+class _PruneCommitUncertainError(RuntimeError):
+ """The database did not acknowledge whether a prune pass committed."""
+
+
+class _PruneRollbackFailedError(RuntimeError):
+ """A prune transaction could not be rolled back safely."""
+
+
+def _run_prune_pass(
+ cutoff: datetime,
+ tables: ShadowTables,
+ after_id: int = 0,
+ max_prune: int | None = None,
) -> dict[str, Any]:
"""One SERIALIZABLE pass over a single id-ordered window of candidate
transactions starting after ``after_id``. The caller wraps this in
@@ -364,44 +415,65 @@ def _run_prune_pass(
# parents: children and the M2M association live on independent
# validity lifecycles and may anchor a still-live row at an older
# transaction than the parent's current live row.
- live_bearing_tables: list[sa.Table] = [*tables.parent, *tables.child]
- if tables.m2m is not None:
- live_bearing_tables.append(tables.m2m)
+ live_bearing_tables: list[sa.Table] = _live_bearing_tables(tables)
# The Celery task runs outside the request-bound DB session, so we
# use a fresh connection rather than ``db.session`` to avoid stepping
# on web-request state.
- with (
- db.engine.connect().execution_options(isolation_level="SERIALIZABLE")
as conn,
- conn.begin(),
- ):
- window = _resolve_prune_window(
- conn, cutoff, live_bearing_tables, after_id, _MAX_PRUNE_BATCH
- )
- tx_ids = window.prunable
-
- parent_rows = _delete_for_transactions(conn, tables.parent, tx_ids)
- child_rows = _delete_for_transactions(conn, tables.child, tx_ids)
- m2m_rows = (
- _delete_for_transactions(conn, [tables.m2m], tx_ids)
- if tables.m2m is not None
- else 0
- )
-
- # Drop the version_transaction rows themselves. ON DELETE
- # CASCADE on version_changes.transaction_id removes the
- # associated change records automatically. Same SQLite bind-
- # parameter chunking applies as the shadow deletes above.
- tx_rows = 0
- for chunk in _chunked(tx_ids, _TX_ID_CHUNK_SIZE):
- tx_rows += (
- conn.execute(
- sa.delete(tables.transaction).where(
- tables.transaction.c.id.in_(chunk)
- )
- ).rowcount
- or 0
+ with db.engine.connect().execution_options(isolation_level="SERIALIZABLE")
as conn:
+ transaction: sa.engine.Transaction = conn.begin()
+ try:
+ window: _PruneWindow = _resolve_prune_window(
+ conn, cutoff, live_bearing_tables, after_id, _MAX_PRUNE_BATCH
)
+ tx_ids: list[int] = window.prunable[:max_prune]
+
+ parent_rows: int = _delete_for_transactions(conn, tables.parent,
tx_ids)
+ child_rows: int = _delete_for_transactions(conn, tables.child,
tx_ids)
+ m2m_rows: int = (
+ _delete_for_transactions(conn, [tables.m2m], tx_ids)
+ if tables.m2m is not None
+ else 0
+ )
+
+ # Drop the version_transaction rows themselves. ON DELETE
+ # CASCADE on version_changes.transaction_id removes the
+ # associated change records automatically. Same SQLite bind-
+ # parameter chunking applies as the shadow deletes above.
+ tx_rows: int = 0
+ for chunk in _chunked(tx_ids, _TX_ID_CHUNK_SIZE):
+ tx_rows += (
+ conn.execute(
+ sa.delete(tables.transaction).where(
+ tables.transaction.c.id.in_(chunk)
+ )
+ ).rowcount
+ or 0
+ )
+ except Exception:
+ try:
+ transaction.rollback() # pylint:
disable=consider-using-transaction
+ except Exception as exc:
+ raise _PruneRollbackFailedError(
+ "version prune rollback failed"
+ ) from exc
+ raise
+ try:
+ transaction.commit() # pylint: disable=consider-using-transaction
+ except Exception as exc:
+ if isinstance(exc, DBAPIError) and not exc.connection_invalidated:
+ try:
+ transaction.rollback() # pylint:
disable=consider-using-transaction
+ except Exception as rollback_exc:
+ raise _PruneRollbackFailedError(
+ "version prune rollback failed"
+ ) from rollback_exc
+ raise
+ # A lost acknowledgement can follow a successful commit. Replaying
+ # the same cursor could prune beyond the per-run cap.
+ raise _PruneCommitUncertainError(
+ "version prune commit outcome unknown"
+ ) from exc
return {
"cutoff": cutoff.isoformat(),
@@ -415,24 +487,27 @@ def _run_prune_pass(
def _run_pass_with_retry(
- cutoff: datetime, tables: ShadowTables, after_id: int
+ cutoff: datetime,
+ tables: ShadowTables,
+ after_id: int,
+ max_prune: int | None = None,
) -> tuple[dict[str, Any], int]:
"""Run one window pass, retrying on serialization conflict. Returns
``(stats, retries_used)``; re-raises the ``OperationalError`` if all
``_MAX_RETRY_ATTEMPTS`` attempts conflict.
Postgres surfaces conflicts as ``SerializationFailure`` (a subclass
- of ``sqlalchemy.exc.OperationalError``). The catch is deliberately the
- broader ``OperationalError`` — it also covers transient faults such as
- SQLite's "database is locked" and dropped connections, all of which are
- safe to retry because each pass is idempotent and runs in its own fresh
- transaction. Without the inline retry a single conflict pushes the next
- attempt 24h out (daily Celery beat), so under sustained write pressure
- the prune could silently fail for days in a row.
+ of ``sqlalchemy.exc.OperationalError``). Definitive errors may retry
+ from a fresh transaction. A commit acknowledgement failure instead raises
+ ``_PruneCommitUncertainError`` and defers the rest of this invocation,
because
+ replaying the same cursor could exceed the per-run cap.
"""
for attempt in range(1, _MAX_RETRY_ATTEMPTS + 1):
try:
- return _run_prune_pass(cutoff, tables, after_id), attempt - 1
+ return (
+ _run_prune_pass(cutoff, tables, after_id, max_prune=max_prune),
+ attempt - 1,
+ )
except OperationalError as exc:
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.retried")
if attempt == _MAX_RETRY_ATTEMPTS:
@@ -455,7 +530,76 @@ def _run_pass_with_retry(
raise AssertionError("unreachable") # pragma: no cover
-def _prune_old_versions_impl(retention_days: int) -> dict[str, Any]:
+def _preview_prune(
+ cutoff: datetime, tables: ShadowTables, max_per_run: int | None
+) -> dict[str, Any]:
+ """Report the full prunable backlog without deleting transactions."""
+ backlog: int = _count_prunable(cutoff, tables)
+ estimated_runs: int = (
+ (backlog + max_per_run - 1) // max_per_run
+ if max_per_run is not None
+ else int(backlog > 0)
+ )
+ stats: dict[str, Any] = {
+ "dry_run": 1,
+ "cutoff": cutoff.isoformat(),
+ "eligible_backlog": backlog,
+ "max_per_run": max_per_run,
+ "estimated_capped_runs": estimated_runs,
+ }
+ logger.info("version_history_retention: DRY RUN %s", stats)
+ return stats
+
+
+def _add_prune_cap_stats(
+ stats: dict[str, Any],
+ cutoff: datetime,
+ tables: ShadowTables,
+ max_per_run: int | None,
+ after_id: int,
+) -> None:
+ """Probe bounded live-run backlog without masking committed deletions."""
+ if max_per_run is None:
+ return
+ cap_reached: bool = stats["pruned_transactions"] >= max_per_run
+ remaining_eligible: int | None = 0
+ count_complete: bool = True
+ if cap_reached:
+ try:
+ remaining_eligible, count_complete = _probe_prunable(
+ cutoff, tables, after_id
+ )
+ except Exception: # pylint: disable=broad-except
+ logger.warning(
+ "version_history_retention: remainder probe failed after
prune",
+ exc_info=True,
+ )
+ stats_logger_manager.instance.incr(
+ f"{_METRIC_PREFIX}.remainder_probe_failed"
+ )
+ remaining_eligible = None
+ count_complete = False
+ stats.update(
+ max_per_run=max_per_run,
+ cap_reached=cap_reached,
+ remaining_eligible=remaining_eligible,
+ remaining_count_complete=count_complete,
+ )
+ if remaining_eligible is not None:
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.remaining_eligible_at_least", remaining_eligible
+ )
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.remaining_count_complete", int(count_complete)
+ )
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.cap_reached", int(cap_reached)
+ )
+
+
+def _prune_old_versions_impl(
+ retention_days: int, max_per_run: int | None = None, dry_run: bool = False
+) -> dict[str, Any]:
"""Pure-Python implementation of the prune. Split out from the
Celery task wrapper so unit tests can call it directly without the
Celery harness.
@@ -481,6 +625,9 @@ def _prune_old_versions_impl(retention_days: int) ->
dict[str, Any]:
24 hours out (daily Celery beat), and under sustained write
pressure the prune can silently fail for many days in a row.
"""
+ max_per_run = validate_retention_cap(
+ max_per_run, "VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN"
+ )
if retention_days == 0 or retention_days < -1:
logger.info(
"version_history_retention: retention disabled or invalid;
skipping",
@@ -502,6 +649,9 @@ def _prune_old_versions_impl(retention_days: int) ->
dict[str, Any]:
if retention_days != -1:
cutoff -= timedelta(days=retention_days)
+ if dry_run:
+ return _preview_prune(cutoff, tables, max_per_run)
+
# Drain the backlog one bounded, id-ordered window at a time. Each
# window is its own retried SERIALIZABLE pass, so memory and
# lock/transaction-hold time stay bounded per pass even on the first
@@ -513,13 +663,24 @@ def _prune_old_versions_impl(retention_days: int) ->
dict[str, Any]:
"pruned_child_shadows": 0,
"pruned_m2m_shadows": 0,
}
- total_retried = 0
- after_id = 0
+ total_retried: int = 0
+ after_id: int = 0
+ pass_stats: dict[str, Any]
+ retries: int
while True:
- pass_stats, retries = _run_pass_with_retry(cutoff, tables, after_id)
+ remaining_budget: int | None = (
+ max_per_run - totals["pruned_transactions"]
+ if max_per_run is not None
+ else None
+ )
+ pass_stats, retries = _run_pass_with_retry(
+ cutoff, tables, after_id, max_prune=remaining_budget
+ )
total_retried += retries
for key in totals:
totals[key] += pass_stats.get(key, 0)
+ if max_per_run is not None and totals["pruned_transactions"] >=
max_per_run:
+ break
if pass_stats.get("candidate_count", 0) < _MAX_PRUNE_BATCH:
break
after_id = pass_stats.get("max_candidate_id", after_id)
@@ -527,6 +688,7 @@ def _prune_old_versions_impl(retention_days: int) ->
dict[str, Any]:
stats: dict[str, Any] = {"cutoff": cutoff.isoformat(), **totals}
if total_retried:
stats["retried"] = total_retried
+ _add_prune_cap_stats(stats, cutoff, tables, max_per_run, after_id)
stats_logger_manager.instance.gauge(
f"{_METRIC_PREFIX}.pruned_transactions", stats["pruned_transactions"]
)
@@ -556,7 +718,27 @@ def prune_old_versions() -> dict[str, Any]:
retention_days: int = _resolve_version_history_retention_days(
configured, legacy_configured, seed=_version_history_retention_seed
)
- return _prune_old_versions_impl(retention_days)
+ try:
+ max_per_run: int | None = validate_retention_cap(
+ current_app.config.get(
+ "VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN", 1000
+ ),
+ "VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN",
+ )
+ except ValueError:
+ logger.warning("version_history_retention: invalid prune cap;
skipping")
+
stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped_invalid_cap")
+ return {"skipped_invalid_cap": 1}
+ dry_run: object =
current_app.config.get("VERSION_HISTORY_PRUNE_DRY_RUN", False)
+ if not isinstance(dry_run, bool):
+ logger.warning("version_history_retention: invalid prune dry-run;
skipping")
+ stats_logger_manager.instance.incr(
+ f"{_METRIC_PREFIX}.skipped_invalid_dry_run"
+ )
+ return {"skipped_invalid_dry_run": 1}
+ return _prune_old_versions_impl(
+ retention_days, max_per_run=max_per_run, dry_run=dry_run
+ )
except Exception: # pylint: disable=broad-except
logger.exception("version_history.prune_old_versions: task failed")
# Emit a failure counter so a prune that fails every night (e.g. an
diff --git a/tests/integration_tests/versioning/retention_prune_tests.py
b/tests/integration_tests/versioning/retention_prune_tests.py
index d870ebf8a4f..5a29a5f7f80 100644
--- a/tests/integration_tests/versioning/retention_prune_tests.py
+++ b/tests/integration_tests/versioning/retention_prune_tests.py
@@ -194,6 +194,71 @@ class TestDashboardVersionRetention(SupersetTestCase):
dashboard.dashboard_title = original_title
db.session.commit()
+ def test_cap_preserves_whole_transactions_across_runs(self) -> None:
+ """Two capped passes each remove two old transactions, never live
rows."""
+ from datetime import datetime, timedelta, timezone
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+ original_title: str = dashboard.dashboard_title
+ try:
+ index: int
+ for index in range(4):
+ dashboard.dashboard_title = f"USA Births Names capped {index}"
+ db.session.commit()
+
+ tx_table: sa.Table = versioning_manager.transaction_cls.__table__
+ with db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+
issued_at=datetime.now(timezone.utc).replace(tzinfo=None)
+ - timedelta(days=100)
+ )
+ )
+
+ count_before: int = db.session.scalar(
+ sa.select(sa.func.count()).select_from(tx_table)
+ )
+ preview: dict[str, Any] = _prune_old_versions_impl(
+ 30, max_per_run=2, dry_run=True
+ )
+ assert preview["eligible_backlog"] >= 4
+ assert preview["estimated_capped_runs"] >= 2
+ assert (
+
db.session.scalar(sa.select(sa.func.count()).select_from(tx_table))
+ == count_before
+ )
+
+ first: dict[str, Any] = _prune_old_versions_impl(30, max_per_run=2)
+ second: dict[str, Any] = _prune_old_versions_impl(30,
max_per_run=2)
+ assert first["pruned_transactions"] == 2
+ assert second["pruned_transactions"] == 2
+ assert first["cap_reached"] is True
+ assert second["cap_reached"] is True
+ assert first["remaining_eligible"] == preview["eligible_backlog"]
- 2
+ assert second["remaining_eligible"] == preview["eligible_backlog"]
- 4
+ assert (
+
db.session.scalar(sa.select(sa.func.count()).select_from(tx_table))
+ == count_before - 4
+ )
+ db.session.expire_all()
+ assert any(
+ row.end_transaction_id is None for row in
_get_version_rows(dashboard)
+ )
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
def test_retention_preserves_live_child_and_m2m_rows(self) -> None:
"""Regression for the live-row preservation BLOCKER.
diff --git
a/tests/unit_tests/commands/deletion_retention/test_purge_chart_detach.py
b/tests/unit_tests/commands/deletion_retention/test_purge_chart_detach.py
index 5f1082c46eb..ca80bf36d0c 100644
--- a/tests/unit_tests/commands/deletion_retention/test_purge_chart_detach.py
+++ b/tests/unit_tests/commands/deletion_retention/test_purge_chart_detach.py
@@ -280,24 +280,22 @@ def test_failed_purge_rolls_back_chart_detach(
def test_dry_run_purge_changes_no_chart(session: Session, app_context: None)
-> None:
"""A dry run counts eligible datasets and never touches their charts."""
- from superset.tasks.deletion_retention import _purge_model
+ from superset.tasks.deletion_retention import _purge_model,
_PurgeModelResult
dataset: SqlaTable
orphan: Slice
dataset, orphan, _, _ = _trashed_dataset_with_charts(session)
before: tuple[Any, ...] = (orphan.datasource_id, orphan.perm,
orphan.schema_perm)
- purged: int
- would_purge: int
- failures: int
- blocked: int
- scan_failures: int
- purged, would_purge, failures, blocked, scan_failures = _purge_model(
- SqlaTable, datetime(2021, 1, 1), True
- )
+ result: _PurgeModelResult = _purge_model(SqlaTable, datetime(2021, 1, 1),
True)
session.expire_all()
- assert (purged, would_purge, failures, blocked, scan_failures) == (0, 1,
0, 0, 0)
+ assert result.purged == 0
+ assert result.would_purge == 1
+ assert result.failures == 0
+ assert result.blocked == 0
+ assert result.scan_failures == 0
+ assert result.commit_uncertain is False
chart: Slice = session.get(Slice, orphan.id)
assert (chart.datasource_id, chart.perm, chart.schema_perm) == before
assert session.get(SqlaTable, dataset.id) is not None
diff --git a/tests/unit_tests/tasks/test_deletion_retention.py
b/tests/unit_tests/tasks/test_deletion_retention.py
index f15a7357dd7..f948594bbce 100644
--- a/tests/unit_tests/tasks/test_deletion_retention.py
+++ b/tests/unit_tests/tasks/test_deletion_retention.py
@@ -24,13 +24,14 @@ import runpy
from collections.abc import Iterator
from datetime import datetime, timedelta
from pathlib import Path
-from typing import Any, ClassVar
+from typing import Any, cast, ClassVar
from unittest.mock import ANY, call, MagicMock, patch
from uuid import UUID, uuid4
import pytest
import sqlalchemy as sa
from flask.config import Config
+from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Mapped, mapped_column, registry, Session
@@ -39,6 +40,7 @@ def app_config(app_context: None, monkeypatch:
pytest.MonkeyPatch) -> Config:
from flask import current_app
current_app.config["SOFT_DELETE_RETENTION_DAYS"] = 30
+ current_app.config["SOFT_DELETE_PURGE_MAX_PER_RUN"] = 0
monkeypatch.setitem(current_app.config, "SOFT_DELETE_RETENTION_DAYS_FUNC",
None)
return current_app.config
@@ -230,13 +232,17 @@ def test_clock_uses_now_not_utcnow() -> None:
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, 0)) as
purge,
+ patch.object(
+ mod, "_purge_model", return_value=mod._PurgeModelResult(0, 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)
+ purge.assert_called_once_with(
+ Slice, now - timedelta(days=30), False, max_per_run=None
+ )
@pytest.mark.parametrize("dry_run", [False, True])
@@ -354,12 +360,17 @@ def
test_scan_failure_keeps_the_counts_earned_before_it(app_context: None) -> No
patch.object(mod, "_iter_eligible_ids", side_effect=pages),
patch.object(mod, "_purge_one", return_value=purged_result),
):
- purged, would, failures, blocked, scan_failures = mod._purge_model(
+ counts: mod._PurgeModelResult = mod._purge_model(
Slice, datetime.now(), dry_run=False
)
- assert (purged, would, failures, blocked) == (2, 0, 0, 0)
- assert scan_failures == 1
+ assert (counts.purged, counts.would_purge, counts.failures,
counts.blocked) == (
+ 2,
+ 0,
+ 0,
+ 0,
+ )
+ assert counts.scan_failures == 1
def test_every_root_failing_reports_the_pass_as_failed(
@@ -376,8 +387,8 @@ def test_every_root_failing_reports_the_pass_as_failed(
monkeypatch.setitem(app_config, "SOFT_DELETE_PURGE_DRY_RUN", False)
- def every_scan_fails(*args: Any, **kwargs: Any) -> tuple[int, int, int,
int, int]:
- return (0, 0, 0, 0, 1)
+ def every_scan_fails(*args: Any, **kwargs: Any) -> task._PurgeModelResult:
+ return task._PurgeModelResult(0, 0, 0, 0, 1)
counter: MagicMock
with (
@@ -429,7 +440,9 @@ def test_root_without_a_table_name_does_not_abort_the_run(
counter: MagicMock
with (
- patch.object(task, "_purge_model", return_value=(1, 0, 0, 0, 0)),
+ patch.object(
+ task, "_purge_model", return_value=task._PurgeModelResult(1, 0, 0,
0, 0)
+ ),
patch.object(task.audit, "reconcile_pending"),
patch.object(task, "resolve_retention_window", return_value=30),
patch.object(
@@ -483,7 +496,11 @@ def
test_unsupported_model_does_not_prevent_supported_models_from_purging(
patch.object(
task,
"_purge_model",
- return_value=(0, 1, 0, 0, 0) if dry_run else (1, 0, 0, 0, 0),
+ return_value=(
+ task._PurgeModelResult(0, 1, 0, 0, 0)
+ if dry_run
+ else task._PurgeModelResult(1, 0, 0, 0, 0)
+ ),
) as purge,
patch.object(task.audit, "reconcile_pending"),
patch.object(task, "resolve_retention_window", return_value=30),
@@ -498,7 +515,7 @@ def
test_unsupported_model_does_not_prevent_supported_models_from_purging(
assert purge.call_count == len(supported_models)
model: type[SoftDeleteMixin]
for model in supported_models:
- assert call(model, ANY, dry_run) in purge.call_args_list
+ assert call(model, ANY, dry_run, max_per_run=None) in
purge.call_args_list
assert result["would_purge" if dry_run else "purged"] == {
"dashboards": 1,
"slices": 1,
@@ -601,11 +618,396 @@ def
test_purge_model_counts_only_committed_deletions(app_context: None) -> None:
patch.object(mod, "_iter_eligible_ids", return_value=[[1]]),
patch.object(mod, "_purge_one", return_value=lost_race),
):
- result: tuple[int, int, int, int, int] = mod._purge_model(
+ result: mod._PurgeModelResult = mod._purge_model(
Slice, datetime.now(), dry_run=False
)
- assert result == (0, 0, 0, 0, 0)
+ assert result == mod._PurgeModelResult(0, 0, 0, 0, 0)
+
+
+def test_purge_cap_skips_blocked_and_failed_roots(app_context: None) -> None:
+ """Blocked and failed attempts leave budget for two successful roots."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.commands.deletion_retention.purge_policy import BlockerReason
+ from superset.models.slice import Slice
+
+ blocked: CascadeResult = CascadeResult(
+ purged=False,
+ entity_type="chart",
+ entity_uuid="blocked",
+ blocker=BlockerReason("referenced", "still referenced"),
+ )
+ confirmed: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="confirmed"
+ )
+ purge_one: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2, 3, 4]]),
+ patch.object(
+ task,
+ "_purge_one",
+ side_effect=[blocked, RuntimeError("cascade failed"), confirmed,
confirmed],
+ ) as purge_one,
+ ):
+ counts: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=2
+ )
+
+ assert (
+ counts.purged,
+ counts.would_purge,
+ counts.failures,
+ counts.blocked,
+ ) == (2, 0, 1, 1)
+ assert [entry.args[1] for entry in purge_one.call_args_list] == [1, 2, 3,
4]
+
+
+def test_purge_model_returns_named_counts(app_context: None) -> None:
+ """Callers can read each purge count without depending on tuple order."""
+ import superset.tasks.deletion_retention as task
+ from superset.models.slice import Slice
+
+ with patch.object(task, "_iter_eligible_ids", return_value=[]):
+ counts: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False
+ )
+
+ assert counts.purged == 0
+ assert counts.would_purge == 0
+ assert counts.failures == 0
+ assert counts.blocked == 0
+ assert counts.scan_failures == 0
+
+
[email protected](
+ "commit_error",
+ [
+ RuntimeError("acknowledgement lost"),
+ OperationalError(
+ "COMMIT", {}, Exception("connection lost"),
connection_invalidated=True
+ ),
+ ],
+)
+def test_uncertain_purge_commit_preserves_pending_audit(
+ app_context: None, commit_error: Exception
+) -> None:
+ """A lost commit acknowledgement cannot be recorded as a failed purge."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+
+ entity: MagicMock = MagicMock(id=1)
+ result: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="uncertain"
+ )
+ audit_fail: MagicMock
+ audit_confirm: MagicMock
+ with (
+ patch.object(task, "skip_visibility_filter"),
+ patch.object(task, "entity_uuid", return_value="uuid-1"),
+ patch.object(task, "dashboard_slice_count", return_value=0),
+ patch.object(task, "suppress_purge_association_versions"),
+ patch.object(task, "cascade_hard_delete", return_value=result),
+ patch.object(task.db, "session") as session,
+ patch.object(task.audit, "write_ahead", return_value=uuid4()),
+ patch.object(task.audit, "fail") as audit_fail,
+ patch.object(task.audit, "confirm") as audit_confirm,
+ ):
+ session.get.return_value = entity
+ session.commit.side_effect = commit_error
+ with pytest.raises(RuntimeError, match="commit outcome"):
+ task._purge_one(Slice, 1, datetime.now())
+
+ audit_fail.assert_not_called()
+ audit_confirm.assert_not_called()
+
+
+def test_failed_commit_rollback_stops_remaining_purge_roots(
+ app_context: None,
+) -> None:
+ """An unconfirmed rollback leaves audit pending and stops later roots."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+
+ entity: MagicMock = MagicMock(id=1)
+ cascade_result: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="uncertain"
+ )
+ audit_fail: MagicMock
+ audit_confirm: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2]]),
+ patch.object(task, "skip_visibility_filter"),
+ patch.object(task, "entity_uuid", return_value="uuid-1"),
+ patch.object(task, "dashboard_slice_count", return_value=0),
+ patch.object(task, "suppress_purge_association_versions"),
+ patch.object(task, "cascade_hard_delete", return_value=cascade_result),
+ patch.object(task.db, "session") as session,
+ patch.object(task.audit, "write_ahead", return_value=uuid4()),
+ patch.object(task.audit, "fail") as audit_fail,
+ patch.object(task.audit, "confirm") as audit_confirm,
+ ):
+ session.get.return_value = entity
+ session.commit.side_effect = OperationalError(
+ "COMMIT", {}, Exception("transaction rejected")
+ )
+ session.rollback.side_effect = [None, RuntimeError("rollback failed")]
+ result: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=2
+ )
+
+ assert result.commit_uncertain is True
+ assert result.failures == 0
+ assert result.purged == 0
+ assert session.get.call_count == 2
+ assert session.commit.call_count == 1
+ assert session.rollback.call_count == 2
+ audit_fail.assert_not_called()
+ audit_confirm.assert_not_called()
+
+
[email protected]("failure_stage", ["flush", "commit"])
+def test_definitive_purge_db_failure_finalizes_audit(
+ app_context: None, failure_stage: str
+) -> None:
+ """An acknowledged database rejection is a failed root, not uncertainty."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+
+ entity: MagicMock = MagicMock(id=1)
+ result: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="rejected"
+ )
+ db_error: OperationalError = OperationalError(
+ failure_stage, {}, Exception("transaction rejected")
+ )
+ audit_fail: MagicMock
+ with (
+ patch.object(task, "skip_visibility_filter"),
+ patch.object(task, "entity_uuid", return_value="uuid-1"),
+ patch.object(task, "dashboard_slice_count", return_value=0),
+ patch.object(task, "suppress_purge_association_versions"),
+ patch.object(task, "cascade_hard_delete", return_value=result),
+ patch.object(task.db, "session") as session,
+ patch.object(task.audit, "write_ahead", return_value=uuid4()),
+ patch.object(task.audit, "fail") as audit_fail,
+ ):
+ session.get.return_value = entity
+ getattr(session, failure_stage).side_effect = db_error
+ with pytest.raises(OperationalError):
+ task._purge_one(Slice, 1, datetime.now())
+
+ audit_fail.assert_called_once()
+ session.rollback.assert_called()
+ if failure_stage == "flush":
+ session.commit.assert_not_called()
+
+
+def test_uncertain_purge_commit_stops_other_roots_and_reserves_cap(
+ app_context: None,
+) -> None:
+ """A possible committed root spends budget and defers the invocation."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ confirmed: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="confirmed"
+ )
+ purge_one: MagicMock
+ count_eligible: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2]]),
+ patch.object(
+ task,
+ "_purge_one",
+ side_effect=[
+ confirmed,
+ task._PurgeCommitUncertainError("commit outcome unknown"),
+ ],
+ ) as purge_one,
+ patch.object(task, "_count_eligible") as count_eligible,
+ patch.object(task.audit, "reconcile_pending"),
+ ):
+ stats: dict[str, Any] = task._purge_impl(30, False, max_per_run=2)
+
+ assert stats["purged"] == {"slices": 1}
+ assert stats["commit_uncertain"] is True
+ assert stats["cap_reached"] is True
+ assert stats["remaining_eligible"] is None
+ assert stats["remaining_count_complete"] is False
+ assert [entry.args[1] for entry in purge_one.call_args_list] == [1, 2]
+ count_eligible.assert_not_called()
+
+
+def test_uncertain_purge_commit_stops_remaining_ids_in_batch(
+ app_context: None,
+) -> None:
+ """A possible committed root stops a model before the next eligible ID."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+
+ confirmed: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="confirmed"
+ )
+ purge_one: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2, 3]]),
+ patch.object(
+ task,
+ "_purge_one",
+ side_effect=[
+ confirmed,
+ task._PurgeCommitUncertainError("commit outcome unknown"),
+ confirmed,
+ ],
+ ) as purge_one,
+ ):
+ counts: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=2
+ )
+
+ assert counts.purged == 1
+ assert counts.commit_uncertain is True
+ assert [entry.args[1] for entry in purge_one.call_args_list] == [1, 2]
+
+
+def test_uncertain_purge_commit_stops_next_batch_with_budget_remaining(
+ app_context: None,
+) -> None:
+ """An uncertain commit must not spend the unused cap on a later page."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+
+ confirmed: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="confirmed"
+ )
+ purge_one: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2], [3]]),
+ patch.object(
+ task,
+ "_purge_one",
+ side_effect=[
+ confirmed,
+ task._PurgeCommitUncertainError("commit outcome unknown"),
+ confirmed,
+ ],
+ ) as purge_one,
+ ):
+ counts: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=3
+ )
+
+ assert counts.purged == 1
+ assert counts.commit_uncertain is True
+ assert [entry.args[1] for entry in purge_one.call_args_list] == [1, 2]
+
+
+def test_uncertain_purge_commit_stops_next_model_with_budget_remaining(
+ app_context: None,
+) -> None:
+ """An unknown outcome halts the run even when the cap is not exhausted."""
+ import superset.tasks.deletion_retention as task
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ purge_model: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(
+ task,
+ "_purge_model",
+ side_effect=[
+ task._PurgeModelResult(
+ purged=0,
+ would_purge=0,
+ failures=0,
+ blocked=0,
+ scan_failures=0,
+ commit_uncertain=True,
+ ),
+ task._PurgeModelResult(
+ purged=1, would_purge=0, failures=0, blocked=0,
scan_failures=0
+ ),
+ ],
+ ) as purge_model,
+ ):
+ scan: task._PurgeScan = task._scan_purge_models(
+ datetime.now(), dry_run=False, max_per_run=5
+ )
+
+ assert scan.commit_uncertain is True
+ assert scan.remaining_budget == 4
+ assert purge_model.call_count == 1
+
+
+def test_post_commit_audit_exception_still_spends_cap(app_context: None) ->
None:
+ """A committed deletion retains its cap slot if audit finalization
raises."""
+ import superset.tasks.deletion_retention as task
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+
+ entity: MagicMock = MagicMock(id=1)
+ result: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="committed"
+ )
+ cascade: MagicMock
+ audit_fail: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2]]),
+ patch.object(task, "skip_visibility_filter"),
+ patch.object(task, "entity_uuid", return_value="uuid-1"),
+ patch.object(task, "dashboard_slice_count", return_value=0),
+ patch.object(task, "suppress_purge_association_versions"),
+ patch.object(task, "cascade_hard_delete", return_value=result) as
cascade,
+ patch.object(task.db, "session") as session,
+ patch.object(task.audit, "write_ahead", return_value=uuid4()),
+ patch.object(task.audit, "confirm", side_effect=RuntimeError("audit
offline")),
+ patch.object(task.audit, "fail") as audit_fail,
+ ):
+ session.get.return_value = entity
+ counts: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=1
+ )
+
+ assert counts.purged == 1
+ assert counts.failures == 0
+ assert cascade.call_count == 1
+ session.commit.assert_called_once()
+ audit_fail.assert_not_called()
+
+
+def test_dry_run_backlog_is_incomplete_after_scan_failure(app_context: None)
-> None:
+ """A partial model scan cannot yield a definitive dry-run estimate."""
+ import superset.tasks.deletion_retention as task
+ from superset.models.slice import Slice
+
+ scan: task._PurgeScan = task._PurgeScan(
+ purged={},
+ would_purge={"slices": 2},
+ unsupported_models={},
+ failures=0,
+ blocked=0,
+ remaining_budget=3,
+ supported_models=[Slice],
+ scan_failures=1,
+ attempted=1,
+ )
+ with patch.object(task, "_scan_purge_models", return_value=scan):
+ stats: dict[str, Any] = task._purge_impl(30, True, max_per_run=3)
+
+ assert stats["would_purge"] == {"slices": 2}
+ assert stats["eligible_backlog"] is None
+ assert stats["estimated_capped_runs"] is None
+ assert stats["backlog_count_complete"] is False
def test_scheduled_purge_fails_closed_when_write_ahead_fails(
@@ -627,14 +1029,13 @@ def
test_scheduled_purge_fails_closed_when_write_ahead_fails(
patch.object(mod.audit, "write_ahead", return_value=None),
):
session.get.return_value = entity
- result: tuple[int, int, int, int, int] = mod._purge_model(
+ result: mod._PurgeModelResult = 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
+ assert (result.purged, result.would_purge, result.blocked) == (0, 0, 0)
+ assert result.failures == 1
@pytest.mark.parametrize("days", [-1, -2])
@@ -651,13 +1052,15 @@ def test_immediate_cutoff_and_invalid_skip(days: int) ->
None:
with (
patch.object(mod, "datetime") as clock,
patch.object(mod, "_soft_delete_models", return_value=[Slice]) as
models,
- patch.object(mod, "_purge_model", return_value=(0, 0, 0, 0, 0)) as
purge,
+ patch.object(
+ mod, "_purge_model", return_value=mod._PurgeModelResult(0, 0, 0,
0, 0)
+ ) as purge,
patch.object(mod.audit, "reconcile_pending") as reconcile,
):
clock.now.return_value = now
result: dict[str, Any] = mod._purge_impl(days, dry_run=False)
if days == -1:
- purge.assert_called_once_with(Slice, now, False)
+ purge.assert_called_once_with(Slice, now, False, max_per_run=None)
else:
assert result == {"skipped": 1}
models.assert_not_called()
@@ -685,7 +1088,9 @@ def test_standalone_window_bounds_reach_safe_purge_cutoff(
),
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, 0)) as
purge,
+ patch.object(
+ mod, "_purge_model", return_value=mod._PurgeModelResult(0, 0, 0,
0, 0)
+ ) as purge,
patch.object(mod.audit, "reconcile_pending"),
):
clock.now.return_value = now
@@ -694,4 +1099,353 @@ def
test_standalone_window_bounds_reach_safe_purge_cutoff(
assert result == {"skipped": 1}
purge.assert_not_called()
else:
- purge.assert_called_once_with(Slice, now - timedelta(days=expected),
False)
+ purge.assert_called_once_with(
+ Slice, now - timedelta(days=expected), False, max_per_run=None
+ )
+
+
+def test_count_eligible_counts_only_roots_archived_before_cutoff(
+ session: Session, app_context: None
+) -> None:
+ """The backlog count runs its real predicate and agrees with the purge
scan."""
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ table: sa.Table = Slice.__table__
+ table.create(session.get_bind())
+ cutoff: datetime = datetime(2026, 1, 31)
+ # Core inserts keep Slice's ORM listeners out of a predicate-only test.
+ session.execute(
+ sa.insert(table),
+ [
+ {"id": 1, "deleted_at": cutoff - timedelta(days=1)},
+ {"id": 2, "deleted_at": cutoff - timedelta(seconds=1)},
+ {"id": 3, "deleted_at": cutoff},
+ {"id": 4, "deleted_at": cutoff + timedelta(days=1)},
+ {"id": 5, "deleted_at": None},
+ ],
+ )
+ session.commit()
+
+ assert task._count_eligible(Slice, cutoff) == 2
+ assert list(task._iter_eligible_ids(Slice, cutoff, batch=10)) == [[1, 2]]
+
+
+def test_purge_cap_counts_committed_roots_across_batches(app_context: None) ->
None:
+ """A failed or blocked root does not consume the successful-purge
budget."""
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ purged: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="purged"
+ )
+ not_purged: CascadeResult = CascadeResult(
+ purged=False, entity_type="chart", entity_uuid="not-purged"
+ )
+ purge_one: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2], [3,
4]]) as scan,
+ patch.object(
+ task, "_purge_one", side_effect=[not_purged, purged, purged]
+ ) as purge_one,
+ ):
+ result: task._PurgeModelResult = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=2
+ )
+
+ assert result == task._PurgeModelResult(2, 0, 0, 0, 0)
+ assert [call.args[1] for call in purge_one.call_args_list] == [1, 2, 3]
+ scan.assert_called_once()
+
+
+def test_scheduled_purge_forwards_positive_cap(app_config: Config) -> None:
+ """The scheduled entrypoint preserves the operator's positive purge
budget."""
+ from superset.tasks import deletion_retention as task
+
+ app_config["SOFT_DELETE_PURGE_MAX_PER_RUN"] = 2
+ app_config["SOFT_DELETE_RETENTION_DAYS"] = 30
+ app_config["SOFT_DELETE_PURGE_DRY_RUN"] = False
+ purge: MagicMock
+ with (
+ patch.object(task, "_purge_impl", return_value={"purged": 2}) as purge,
+ patch.object(task, "resolve_retention_window", return_value=30),
+ patch.object(
+ task.feature_flag_manager, "is_feature_enabled", return_value=True
+ ),
+ ):
+ result: dict[str, Any] = task.purge_soft_deleted.run()
+
+ purge.assert_called_once_with(30, False, max_per_run=2)
+ assert result == {"purged": 2}
+
+
+def test_scheduled_purge_rejects_invalid_cap_before_work(app_config: Config)
-> None:
+ """Malformed scheduled budgets fail closed instead of removing
everything."""
+ from superset.tasks import deletion_retention as task
+
+ app_config["SOFT_DELETE_PURGE_MAX_PER_RUN"] = -1
+ purge: MagicMock
+ with (
+ patch.object(task, "_purge_impl") as purge,
+ patch.object(
+ task.feature_flag_manager, "is_feature_enabled", return_value=True
+ ),
+ ):
+ result: dict[str, Any] = task.purge_soft_deleted.run()
+
+ assert result == {"skipped_invalid_cap": 1}
+ purge.assert_not_called()
+
+
+def test_purge_budget_spans_models_and_runs(app_context: None) -> None:
+ """A full first run leaves later models for a subsequent invocation."""
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ purge: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(
+ task,
+ "_purge_model",
+ side_effect=[
+ task._PurgeModelResult(2, 0, 0, 0, 0),
+ task._PurgeModelResult(0, 0, 0, 0, 0),
+ task._PurgeModelResult(1, 0, 0, 0, 0),
+ ],
+ ) as purge,
+ patch.object(task, "_count_eligible", side_effect=[0, 3, 0, 2]),
+ patch.object(task.audit, "reconcile_pending"),
+ ):
+ first: dict[str, Any] = task._purge_impl(30, False, max_per_run=2)
+ second: dict[str, Any] = task._purge_impl(30, False, max_per_run=2)
+
+ assert first["purged"] == {"slices": 2}
+ assert first["cap_reached"] is True
+ assert first["remaining_eligible"] == 3
+ assert second["purged"] == {"dashboards": 1}
+ assert second["cap_reached"] is False
+ assert second["remaining_eligible"] == 2
+ assert [entry.args[0] for entry in purge.call_args_list] == [
+ Slice,
+ Slice,
+ Dashboard,
+ ]
+ assert [entry.kwargs["max_per_run"] for entry in purge.call_args_list] == [
+ 2,
+ 2,
+ 2,
+ ]
+
+
+def test_purge_budget_remaining_after_first_model(app_context: None) -> None:
+ """A partially used cap limits the next model's eligible roots."""
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ cutoff: datetime = datetime(2026, 10, 1)
+ scan: task._PurgeScan
+ purge_one: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(
+ task, "_iter_eligible_ids", side_effect=[iter([[1]]), iter([[2,
3]])]
+ ),
+ patch.object(
+ task,
+ "_purge_one",
+ return_value=CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="x"
+ ),
+ ) as purge_one,
+ ):
+ scan = task._scan_purge_models(cutoff, dry_run=False, max_per_run=2)
+
+ assert scan.purged == {"slices": 1, "dashboards": 1}
+ assert scan.remaining_budget == 0
+ assert [entry.args[:2] for entry in purge_one.call_args_list] == [
+ (Slice, 1),
+ (Dashboard, 2),
+ ]
+
+
[email protected]("rollback_fails", [False, True])
+def test_partial_scan_failure_preserves_counts_and_shared_budget(
+ app_context: None,
+ rollback_fails: bool,
+) -> None:
+ """A failed scan/rollback cannot give the next model spent purge budget."""
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ def partial_scan() -> Iterator[list[int]]:
+ """Return a committed root's page, then fail fetching the next page."""
+ yield [1]
+ raise RuntimeError("scan failed")
+
+ purge_one: MagicMock
+ rollback: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(
+ task,
+ "_iter_eligible_ids",
+ side_effect=[partial_scan(), iter([[2, 3]])],
+ ),
+ patch.object(
+ task,
+ "_purge_one",
+ return_value=CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="x"
+ ),
+ ) as purge_one,
+ patch.object(
+ task.db.session,
+ "rollback",
+ side_effect=[RuntimeError("rollback failed"), None]
+ if rollback_fails
+ else None,
+ ) as rollback,
+ patch.object(task.stats_logger_manager.instance, "incr"),
+ ):
+ scan: task._PurgeScan = task._scan_purge_models(
+ datetime(2026, 10, 7), dry_run=False, max_per_run=2
+ )
+
+ assert scan.purged == {"slices": 1, "dashboards": 1}
+ assert scan.scan_failures == 1
+ assert scan.remaining_budget == 0
+ assert [entry.args[:2] for entry in purge_one.call_args_list] == [
+ (Slice, 1),
+ (Dashboard, 2),
+ ]
+ rollback.assert_called_once()
+
+
+def test_purge_model_priority_rotates_across_days() -> None:
+ """A cap smaller than the model count still gives each model first turn."""
+ from superset.connectors.sqla.models import SqlaTable
+ from superset.models.dashboard import Dashboard
+ from superset.models.helpers import SoftDeleteMixin
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ models: list[type[SoftDeleteMixin]] = [Slice, Dashboard, SqlaTable]
+ with patch.object(task, "_soft_delete_models", return_value=models):
+ day: int
+ priorities: set[type[SoftDeleteMixin]] = set()
+ for day in (1, 2, 3):
+ priorities.add(task._ordered_purge_models(datetime(2026, 1,
day))[0])
+
+ assert priorities == set(models)
+
+
[email protected]("cap", [None, 0, 3])
+def test_purge_dry_run_counts_full_backlog_without_writes(
+ app_context: None, cap: int | None
+) -> None:
+ """Dry-run eligibility is independent of a configured deletion budget."""
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ purge_one: MagicMock
+ reconcile: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(
+ task,
+ "_iter_eligible_ids",
+ side_effect=[
+ iter([[1, 2], [3, 4]]),
+ iter([[5, 6, 7]]),
+ ],
+ ),
+ patch.object(task, "_purge_one") as purge_one,
+ patch.object(task.audit, "reconcile_pending") as reconcile,
+ ):
+ result: dict[str, Any] = task._purge_impl(30, True, max_per_run=cap)
+
+ assert result["would_purge"] == {"slices": 4, "dashboards": 3}
+ purge_one.assert_not_called()
+ reconcile.assert_not_called()
+ assert result["eligible_backlog"] == 7
+ assert result["estimated_capped_runs"] == (3 if cap == 3 else 1)
+
+
[email protected]("invalid", [-1, True, "2", 1.5])
+def test_purge_rejects_invalid_cap(invalid: object) -> None:
+ """Invalid budgets cannot silently become an unlimited direct call."""
+ from superset.tasks import deletion_retention as task
+
+ with pytest.raises(ValueError, match="SOFT_DELETE_PURGE_MAX_PER_RUN"):
+ task._purge_impl(30, False, max_per_run=cast(int | None, invalid))
+
+
[email protected]("rollback_fails", [False, True])
+def test_purge_remainder_failure_keeps_committed_totals(
+ app_context: None, rollback_fails: bool
+) -> None:
+ """A post-commit count failure cannot turn successful purges into an
error."""
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ scan: task._PurgeScan = task._PurgeScan(
+ purged={"slices": 2},
+ would_purge={},
+ unsupported_models={},
+ failures=0,
+ blocked=0,
+ remaining_budget=0,
+ supported_models=[Slice],
+ )
+ with (
+ patch.object(task, "_scan_purge_models", return_value=scan),
+ patch.object(task, "_count_eligible", side_effect=RuntimeError("count
failed")),
+ patch.object(task.audit, "reconcile_pending"),
+ patch.object(
+ task.db.session,
+ "rollback",
+ side_effect=RuntimeError("rollback failed") if rollback_fails else
None,
+ ) as rollback,
+ ):
+ result: dict[str, Any] = task._purge_impl(30, False, max_per_run=2)
+
+ assert result["purged"] == {"slices": 2}
+ assert result["cap_reached"] is True
+ assert result["remaining_eligible"] is None
+ assert result["remaining_count_complete"] is False
+ rollback.assert_called_once()
+
+
+def test_purge_remainder_incomplete_after_scan_failure(app_context: None) ->
None:
+ """An unscanned root cannot be silently omitted from a complete count."""
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ scan: task._PurgeScan = task._PurgeScan(
+ purged={},
+ would_purge={},
+ unsupported_models={},
+ failures=0,
+ blocked=0,
+ remaining_budget=1,
+ supported_models=[Slice],
+ scan_failures=1,
+ attempted=2,
+ )
+ with (
+ patch.object(task, "_scan_purge_models", return_value=scan),
+ patch.object(task, "_count_eligible", return_value=0),
+ patch.object(task.audit, "reconcile_pending"),
+ ):
+ result: dict[str, Any] = task._purge_impl(30, False, max_per_run=1)
+
+ assert result["remaining_eligible"] is None
+ assert result["remaining_count_complete"] is False
diff --git a/tests/unit_tests/tasks/test_version_history_retention.py
b/tests/unit_tests/tasks/test_version_history_retention.py
index 8b5b18f38fb..f9790a4220e 100644
--- a/tests/unit_tests/tasks/test_version_history_retention.py
+++ b/tests/unit_tests/tasks/test_version_history_retention.py
@@ -30,10 +30,11 @@ from __future__ import annotations
from collections.abc import Iterator
from datetime import datetime
-from typing import Any
-from unittest.mock import MagicMock, patch
+from typing import Any, cast
+from unittest.mock import call, MagicMock, patch
import pytest
+import sqlalchemy as sa
from flask import Flask
from sqlalchemy.exc import OperationalError
from sqlalchemy_continuum.exc import ClassNotVersioned
@@ -81,7 +82,7 @@ def test_task_reads_canonical_application_retention(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(int(value))
+ prune.assert_called_once_with(int(value), max_per_run=1000, dry_run=False)
stats.incr.assert_not_called()
@@ -103,7 +104,7 @@ def
test_task_preserves_legacy_only_custom_application_config(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(expected)
+ prune.assert_called_once_with(expected, max_per_run=1000, dry_run=False)
@pytest.mark.parametrize(("legacy", "expected"), [(0, 0), (365, 365)])
@@ -124,7 +125,7 @@ def
test_task_keeps_legacy_when_custom_module_imports_new_default(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(expected)
+ prune.assert_called_once_with(expected, max_per_run=1000, dry_run=False)
@pytest.mark.parametrize(("canonical", "legacy"), [(7, 365), (365, 0), (-1,
-1)])
@@ -145,7 +146,7 @@ def
test_task_honors_explicit_canonical_config_with_legacy_key_present(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(canonical)
+ prune.assert_called_once_with(canonical, max_per_run=1000, dry_run=False)
@pytest.mark.parametrize("value", [-1.0, -1.5, True, False, None, "abc",
"30d"])
@@ -163,7 +164,7 @@ def test_task_does_not_coerce_invalid_input_to_immediate(
) as prune,
):
assert version_history_retention.prune_old_versions.run() == {}
- prune.assert_called_once_with(0)
+ prune.assert_called_once_with(0, max_per_run=1000, dry_run=False)
stats.incr.assert_not_called()
assert "Invalid VERSION_HISTORY_RETENTION_DAYS" in caplog.text
@@ -186,7 +187,7 @@ def test_task_env_canonical_precedes_legacy_key(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(30)
+ prune.assert_called_once_with(30, max_per_run=1000, dry_run=False)
def test_task_without_either_key_uses_environment_seed(stats: MagicMock) ->
None:
@@ -201,7 +202,9 @@ def
test_task_without_either_key_uses_environment_seed(stats: MagicMock) -> None
):
assert version_history_retention.prune_old_versions() == {}
prune.assert_called_once_with(
- version_history_retention._version_history_retention_seed
+ version_history_retention._version_history_retention_seed,
+ max_per_run=1000,
+ dry_run=False,
)
@@ -223,7 +226,7 @@ def test_task_defers_invalid_legacy_retention_with_zero(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(0)
+ prune.assert_called_once_with(0, max_per_run=1000, dry_run=False)
stats.incr.assert_not_called()
assert "Invalid SUPERSET_VERSION_HISTORY_RETENTION_DAYS" in caplog.text
@@ -243,7 +246,7 @@ def
test_task_defers_invalid_legacy_only_retention_with_zero(
) as prune,
):
assert version_history_retention.prune_old_versions() == {}
- prune.assert_called_once_with(0)
+ prune.assert_called_once_with(0, max_per_run=1000, dry_run=False)
stats.incr.assert_not_called()
assert "Invalid SUPERSET_VERSION_HISTORY_RETENTION_DAYS" in caplog.text
@@ -263,7 +266,7 @@ def test_task_normalizes_string_retention_config(stats:
MagicMock) -> None:
result = version_history_retention.prune_old_versions()
assert result == {"pruned_transactions": 0}
- prune.assert_called_once_with(30)
+ prune.assert_called_once_with(30, max_per_run=1000, dry_run=False)
stats.incr.assert_not_called()
@@ -407,9 +410,639 @@ def test_immediate_cutoff_and_invalid_skip(stats:
MagicMock, days: int) -> None:
days
)
if days == -1:
- run_pass.assert_called_once_with(now, [], 0)
+ run_pass.assert_called_once_with(now, [], 0, max_prune=None)
assert result["cutoff"] == now.isoformat()
else:
assert result == {"skipped": 1}
tables.assert_not_called()
run_pass.assert_not_called()
+
+
+def test_prune_cap_accounts_only_for_committed_passes(stats: MagicMock) ->
None:
+ """A retried pass uses the same allowance and cannot overshoot the cap."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ first: dict[str, int] = {
+ "candidate_count": 1000,
+ "max_candidate_id": 1000,
+ "pruned_transactions": 2,
+ }
+ second: dict[str, int] = {
+ "candidate_count": 1000,
+ "max_candidate_id": 2000,
+ "pruned_transactions": 1,
+ }
+ run_pass: MagicMock
+ with (
+ patch.object(
+ version_history_retention, "_resolve_shadow_tables",
return_value=tables
+ ),
+ patch.object(
+ version_history_retention,
+ "_run_pass_with_retry",
+ side_effect=[(first, 1), (second, 0)],
+ ) as run_pass,
+ patch.object(
+ version_history_retention, "_probe_prunable", return_value=(4,
False)
+ ),
+ ):
+ result: dict[str, Any] =
version_history_retention._prune_old_versions_impl(
+ retention_days=30, max_per_run=3
+ )
+
+ assert result["pruned_transactions"] == 3
+ assert result["retried"] == 1
+ assert result["cap_reached"] is True
+ assert result["remaining_eligible"] == 4
+ assert result["remaining_count_complete"] is False
+ stats.gauge.assert_any_call(
+ "superset.versioning.retention.remaining_count_complete", 0
+ )
+ assert run_pass.call_args_list[0].kwargs["max_prune"] == 3
+ assert run_pass.call_args_list[1].kwargs["max_prune"] == 1
+
+
+def test_prune_cap_probes_after_preserved_candidate_windows(
+ stats: MagicMock,
+) -> None:
+ """The remainder probe skips windows already scanned before the cap."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ batch_size: int = version_history_retention._MAX_PRUNE_BATCH
+ windows: dict[int, version_history_retention._PruneWindow] = {
+ 0: version_history_retention._PruneWindow([], batch_size, 1000),
+ 1000: version_history_retention._PruneWindow([1500], batch_size, 2000),
+ }
+
+ def resolve_window(
+ _conn: sa.engine.Connection,
+ _cutoff: datetime,
+ _shadow_tables: list[sa.Table],
+ after_id: int,
+ _limit: int,
+ ) -> version_history_retention._PruneWindow:
+ """Return the candidate window at the requested scan cursor."""
+ return windows[after_id]
+
+ resolve: MagicMock
+ with (
+ patch.object(
+ version_history_retention, "_resolve_shadow_tables",
return_value=tables
+ ),
+ patch.object(
+ version_history_retention,
+ "_run_pass_with_retry",
+ side_effect=[
+ ({"candidate_count": batch_size, "max_candidate_id": 1000}, 0),
+ (
+ {
+ "candidate_count": batch_size,
+ "max_candidate_id": 2000,
+ "pruned_transactions": 1,
+ },
+ 0,
+ ),
+ ],
+ ),
+ patch.object(version_history_retention, "db"),
+ patch.object(
+ version_history_retention,
+ "_resolve_prune_window",
+ side_effect=resolve_window,
+ ) as resolve,
+ ):
+ result: dict[str, Any] =
version_history_retention._prune_old_versions_impl(
+ retention_days=30, max_per_run=1
+ )
+
+ assert result["pruned_transactions"] == 1
+ assert result["remaining_eligible"] == 1
+ assert result["remaining_count_complete"] is False
+ assert resolve.call_args.args[3] == 1000
+ stats.gauge.assert_any_call(
+ "superset.versioning.retention.remaining_eligible_at_least", 1
+ )
+
+
[email protected]("cap", [None, 0, 3])
+def test_prune_dry_run_counts_all_eligible_without_writes(
+ stats: MagicMock, cap: int | None
+) -> None:
+ """A dry run reports the whole backlog and capped-run estimate."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ run_pass: MagicMock
+ with (
+ patch.object(
+ version_history_retention, "_resolve_shadow_tables",
return_value=tables
+ ),
+ patch.object(version_history_retention, "_count_prunable",
return_value=7),
+ patch.object(version_history_retention, "_run_pass_with_retry") as
run_pass,
+ ):
+ result: dict[str, Any] =
version_history_retention._prune_old_versions_impl(
+ 30, max_per_run=cap, dry_run=True
+ )
+
+ assert result["eligible_backlog"] == 7
+ assert result["estimated_capped_runs"] == (3 if cap == 3 else 1)
+ run_pass.assert_not_called()
+ stats.gauge.assert_not_called()
+
+
[email protected]("helper_name", ["_count_prunable", "_probe_prunable"])
+def test_remainder_reads_use_default_isolation(helper_name: str) -> None:
+ """Backlog measurement does not request the delete pass's isolation."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ engine: MagicMock = MagicMock()
+ window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[], candidate_count=0, max_candidate_id=0
+ )
+ )
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention, "_resolve_prune_window",
return_value=window
+ ),
+ ):
+ mock_db.engine = engine
+ result: int | tuple[int, bool] = getattr(
+ version_history_retention, helper_name
+ )(datetime(2026, 1, 1), tables)
+
+ assert result in (0, (0, True))
+ engine.connect.return_value.execution_options.assert_not_called()
+
+
+def test_dry_run_count_accumulates_windows_and_closes_read_transactions() ->
None:
+ """A paged backlog count sums windows and releases each read snapshot."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ engine: MagicMock = MagicMock()
+ full_window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1, 2],
+ candidate_count=version_history_retention._MAX_PRUNE_BATCH,
+ max_candidate_id=1000,
+ )
+ )
+ final_window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1001], candidate_count=1, max_candidate_id=1001
+ )
+ )
+ mock_db: MagicMock
+ resolve: MagicMock
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention,
+ "_resolve_prune_window",
+ side_effect=[full_window, final_window],
+ ) as resolve,
+ ):
+ mock_db.engine = engine
+ count: int = version_history_retention._count_prunable(
+ datetime(2026, 1, 1), tables
+ )
+
+ assert count == 3
+ assert resolve.call_args_list[0].args[3] == 0
+ assert resolve.call_args_list[1].args[3] == full_window.max_candidate_id
+ assert engine.connect.call_count == 2
+ assert engine.connect.return_value.__exit__.call_count == 2
+ first_exit: int = engine.mock_calls.index(call.connect().__exit__(None,
None, None))
+ second_connect: int = engine.mock_calls.index(call.connect(), first_exit +
1)
+ assert first_exit < second_connect
+
+
[email protected]("invalid", [-1, True, "3", 2.5])
+def test_prune_rejects_invalid_cap_before_work(invalid: object) -> None:
+ """Malformed budgets fail closed before resolving any shadow table."""
+ with (
+ patch.object(version_history_retention, "_resolve_shadow_tables") as
resolve,
+ pytest.raises(ValueError, match="VERSION_HISTORY_PRUNE_MAX"),
+ ):
+ version_history_retention._prune_old_versions_impl(
+ 30, max_per_run=cast(int | None, invalid)
+ )
+ resolve.assert_not_called()
+
+
+def test_prune_retry_reuses_the_same_transaction_budget(stats: MagicMock) ->
None:
+ """A rolled-back SERIALIZABLE attempt consumes no transaction allowance."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ cutoff: datetime = datetime(2026, 1, 1)
+ run_pass: MagicMock
+ with (
+ patch.object(
+ version_history_retention,
+ "_run_prune_pass",
+ side_effect=[
+ OperationalError("SELECT 1", {}, Exception("serialization
conflict")),
+ {"pruned_transactions": 2},
+ ],
+ ) as run_pass,
+ patch.object(version_history_retention.time, "sleep"),
+ ):
+ result: tuple[dict[str, Any], int] = (
+ version_history_retention._run_pass_with_retry(
+ cutoff, tables, after_id=4, max_prune=2
+ )
+ )
+
+ assert result == ({"pruned_transactions": 2}, 1)
+ assert run_pass.call_args_list == [
+ call(cutoff, tables, 4, max_prune=2),
+ call(cutoff, tables, 4, max_prune=2),
+ ]
+ stats.incr.assert_called_once_with("superset.versioning.retention.retried")
+
+
[email protected](
+ "commit_error",
+ [
+ OperationalError(
+ "COMMIT", {}, Exception("connection lost"),
connection_invalidated=True
+ ),
+ RuntimeError("acknowledgement lost"),
+ ],
+)
+def test_prune_does_not_retry_an_uncertain_commit(
+ stats: MagicMock, commit_error: Exception
+) -> None:
+ """A lost commit acknowledgement must not spend another prune window."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[],
+ child=[],
+ m2m=None,
+ transaction=sa.table("version_transaction", sa.column("id")),
+ )
+ )
+ window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1], candidate_count=1, max_candidate_id=1
+ )
+ )
+ engine: MagicMock = MagicMock()
+ engine_connection: MagicMock = engine.connect.return_value
+ connection: MagicMock = (
+ engine_connection.execution_options.return_value.__enter__.return_value
+ )
+ transaction: MagicMock = connection.begin.return_value
+ transaction.commit.side_effect = commit_error
+ mock_db: MagicMock
+ sleep: MagicMock
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention, "_resolve_prune_window",
return_value=window
+ ),
+ patch.object(
+ version_history_retention, "_delete_for_transactions",
return_value=0
+ ),
+ patch.object(version_history_retention.time, "sleep") as sleep,
+ ):
+ mock_db.engine = engine
+ with pytest.raises(RuntimeError, match="commit outcome"):
+ version_history_retention._run_pass_with_retry(
+ datetime(2026, 1, 1), tables, after_id=0, max_prune=1
+ )
+
+ assert engine.connect.call_count == 1
+ sleep.assert_not_called()
+ stats.incr.assert_not_called()
+
+
+def test_prune_cap_limits_every_shadow_delete_to_capped_ids() -> None:
+ """Every shadow category and transaction delete uses the same capped
IDs."""
+ parent: sa.Table = sa.table("parent")
+ child: sa.Table = sa.table("child")
+ m2m: sa.Table = sa.table("m2m")
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[parent],
+ child=[child],
+ m2m=m2m,
+ transaction=sa.table("version_transaction", sa.column("id")),
+ )
+ )
+ window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1, 2, 3], candidate_count=3, max_candidate_id=3
+ )
+ )
+ engine: MagicMock = MagicMock()
+ connection_context: MagicMock = (
+ engine.connect.return_value.execution_options.return_value
+ )
+ connection: MagicMock = connection_context.__enter__.return_value
+ connection.execute.return_value.rowcount = 0
+ delete: MagicMock
+ mock_db: MagicMock
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention, "_resolve_prune_window",
return_value=window
+ ),
+ patch.object(
+ version_history_retention, "_delete_for_transactions",
return_value=0
+ ) as delete,
+ ):
+ mock_db.engine = engine
+ result: dict[str, Any] = version_history_retention._run_prune_pass(
+ datetime(2026, 1, 1), tables, after_id=0, max_prune=2
+ )
+
+ assert delete.call_args_list == [
+ call(connection, [parent], [1, 2]),
+ call(connection, [child], [1, 2]),
+ call(connection, [m2m], [1, 2]),
+ ]
+ transaction_delete: sa.sql.dml.Delete =
connection.execute.call_args.args[0]
+ assert transaction_delete.compile().params == {"id_1": [1, 2]}
+ assert result["pruned_transactions"] == 0
+
+
+def test_prune_child_delete_failure_rolls_back_parent_delete() -> None:
+ """A child-delete failure rolls back the pass instead of committing its
parent."""
+ metadata: sa.MetaData = sa.MetaData()
+ parent: sa.Table = sa.Table("parent", metadata)
+ child: sa.Table = sa.Table("child", metadata)
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[parent],
+ child=[child],
+ m2m=sa.Table("m2m", metadata),
+ transaction=sa.Table(
+ "version_transaction", metadata, sa.Column("id", sa.Integer)
+ ),
+ )
+ )
+ window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1], candidate_count=1, max_candidate_id=1
+ )
+ )
+ engine: MagicMock = MagicMock()
+ connection_context: MagicMock = (
+ engine.connect.return_value.execution_options.return_value
+ )
+ connection: MagicMock = connection_context.__enter__.return_value
+ transaction: MagicMock = connection.begin.return_value
+ delete_error: RuntimeError = RuntimeError("child-shadow delete failed")
+ delete: MagicMock
+ mock_db: MagicMock
+ raised: pytest.ExceptionInfo[RuntimeError]
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention, "_resolve_prune_window",
return_value=window
+ ),
+ patch.object(
+ version_history_retention,
+ "_delete_for_transactions",
+ side_effect=[1, delete_error],
+ ) as delete,
+ ):
+ mock_db.engine = engine
+ with pytest.raises(RuntimeError, match="child-shadow delete failed")
as raised:
+ version_history_retention._run_prune_pass(
+ datetime(2026, 1, 1), tables, max_prune=1
+ )
+
+ assert raised.value is delete_error
+ assert delete.call_args_list == [
+ call(connection, [parent], [1]),
+ call(connection, [child], [1]),
+ ]
+ transaction.rollback.assert_called_once_with()
+ transaction.commit.assert_not_called()
+ connection.execute.assert_not_called()
+
+
+def test_prune_retries_definitive_db_commit_failure(stats: MagicMock) -> None:
+ """An acknowledged transaction rejection may retry the same capped
window."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[],
+ child=[],
+ m2m=None,
+ transaction=sa.table("version_transaction", sa.column("id")),
+ )
+ )
+ window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1], candidate_count=1, max_candidate_id=1
+ )
+ )
+ engine: MagicMock = MagicMock()
+ engine_connection: MagicMock = engine.connect.return_value
+ connection: MagicMock = (
+ engine_connection.execution_options.return_value.__enter__.return_value
+ )
+ transaction: MagicMock = connection.begin.return_value
+ commit_error: OperationalError = OperationalError(
+ "COMMIT", {}, Exception("serialization failure")
+ )
+ transaction.commit.side_effect = [commit_error, None]
+ mock_db: MagicMock
+ sleep: MagicMock
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention, "_resolve_prune_window",
return_value=window
+ ),
+ patch.object(
+ version_history_retention, "_delete_for_transactions",
return_value=0
+ ),
+ patch.object(version_history_retention.time, "sleep") as sleep,
+ ):
+ mock_db.engine = engine
+ result: tuple[dict[str, Any], int] = (
+ version_history_retention._run_pass_with_retry(
+ datetime(2026, 1, 1), tables, after_id=0, max_prune=1
+ )
+ )
+
+ assert result[1] == 1
+ assert engine.connect.call_count == 2
+ transaction.rollback.assert_called_once()
+ sleep.assert_called_once()
+ stats.incr.assert_called_once_with("superset.versioning.retention.retried")
+
+
+def test_prune_failed_rollback_stops_retry(stats: MagicMock) -> None:
+ """A failed rollback cannot be classified as a retryable rejection."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[],
+ child=[],
+ m2m=None,
+ transaction=sa.table("version_transaction", sa.column("id")),
+ )
+ )
+ window: version_history_retention._PruneWindow = (
+ version_history_retention._PruneWindow(
+ prunable=[1], candidate_count=1, max_candidate_id=1
+ )
+ )
+ engine: MagicMock = MagicMock()
+ engine_connection: MagicMock = engine.connect.return_value
+ connection: MagicMock = (
+ engine_connection.execution_options.return_value.__enter__.return_value
+ )
+ transaction: MagicMock = connection.begin.return_value
+ transaction.commit.side_effect = OperationalError(
+ "COMMIT", {}, Exception("transaction rejected")
+ )
+ transaction.rollback.side_effect = RuntimeError("rollback failed")
+ mock_db: MagicMock
+ sleep: MagicMock
+ with (
+ patch.object(version_history_retention, "db") as mock_db,
+ patch.object(
+ version_history_retention, "_resolve_prune_window",
return_value=window
+ ),
+ patch.object(
+ version_history_retention, "_delete_for_transactions",
return_value=0
+ ),
+ patch.object(version_history_retention.time, "sleep") as sleep,
+ ):
+ mock_db.engine = engine
+ with pytest.raises(
+ version_history_retention._PruneRollbackFailedError,
+ match="rollback failed",
+ ):
+ version_history_retention._run_pass_with_retry(
+ datetime(2026, 1, 1), tables, after_id=0, max_prune=1
+ )
+
+ assert engine.connect.call_count == 1
+ sleep.assert_not_called()
+ stats.incr.assert_not_called()
+
+
+def test_scheduled_prune_rejects_invalid_cap_before_work(stats: MagicMock) ->
None:
+ """A malformed scheduled budget cannot enter the prune implementation."""
+ app: Flask = Flask(__name__)
+ app.config.update(
+ VERSION_HISTORY_RETENTION_DAYS=30,
+ VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN=-1,
+ )
+ prune: MagicMock
+ with (
+ app.app_context(),
+ patch.object(version_history_retention, "_prune_old_versions_impl") as
prune,
+ ):
+ result: dict[str, Any] = version_history_retention.prune_old_versions()
+
+ assert result == {"skipped_invalid_cap": 1}
+ prune.assert_not_called()
+ stats.incr.assert_called_once_with(
+ "superset.versioning.retention.skipped_invalid_cap"
+ )
+
+
+def test_scheduled_prune_passes_dry_run_and_cap_to_impl() -> None:
+ """The scheduled wrapper preserves both operator-selected controls."""
+ app: Flask = Flask(__name__)
+ app.config.update(
+ VERSION_HISTORY_RETENTION_DAYS=30,
+ VERSION_HISTORY_PRUNE_DRY_RUN=True,
+ VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN=2,
+ )
+ prune: MagicMock
+ with (
+ app.app_context(),
+ patch.object(
+ version_history_retention, "_prune_old_versions_impl",
return_value={}
+ ) as prune,
+ ):
+ result: dict[str, Any] = version_history_retention.prune_old_versions()
+
+ assert result == {}
+ prune.assert_called_once_with(30, max_per_run=2, dry_run=True)
+
+
[email protected]("invalid", [None, 0, 1, "false", [], {}])
+def test_scheduled_prune_skips_invalid_dry_run_before_work(
+ stats: MagicMock, caplog: pytest.LogCaptureFixture, invalid: object
+) -> None:
+ """Malformed dry-run configuration skips without marking the task
failed."""
+ app: Flask = Flask(__name__)
+ app.config.update(
+ VERSION_HISTORY_RETENTION_DAYS=30,
+ VERSION_HISTORY_PRUNE_DRY_RUN=invalid,
+ )
+ prune: MagicMock
+ with (
+ app.app_context(),
+ patch.object(version_history_retention, "_prune_old_versions_impl") as
prune,
+ ):
+ result: dict[str, Any] = version_history_retention.prune_old_versions()
+
+ assert result == {"skipped_invalid_dry_run": 1}
+ prune.assert_not_called()
+ stats.incr.assert_called_once_with(
+ "superset.versioning.retention.skipped_invalid_dry_run"
+ )
+ assert "invalid prune dry-run" in caplog.text
+
+
+def test_remainder_probe_failure_does_not_mask_committed_prune(
+ stats: MagicMock,
+) -> None:
+ """A measurement fault leaves committed deletion totals successful."""
+ tables: version_history_retention.ShadowTables = (
+ version_history_retention.ShadowTables(
+ parent=[], child=[], m2m=None, transaction=MagicMock()
+ )
+ )
+ with (
+ patch.object(
+ version_history_retention, "_resolve_shadow_tables",
return_value=tables
+ ),
+ patch.object(
+ version_history_retention,
+ "_run_pass_with_retry",
+ return_value=({"pruned_transactions": 2}, 0),
+ ),
+ patch.object(
+ version_history_retention,
+ "_probe_prunable",
+ side_effect=OperationalError("SELECT 1", {}, Exception("database
offline")),
+ ),
+ ):
+ result: dict[str, Any] =
version_history_retention._prune_old_versions_impl(
+ 30, max_per_run=2
+ )
+
+ assert result["pruned_transactions"] == 2
+ assert result["cap_reached"] is True
+ assert result["remaining_eligible"] is None
+ assert result["remaining_count_complete"] is False
+ stats.incr.assert_called_once_with(
+ "superset.versioning.retention.remainder_probe_failed"
+ )
