codeant-ai-for-open-source[bot] commented on code in PR #41549: URL: https://github.com/apache/superset/pull/41549#discussion_r3674224887
########## superset/tasks/deletion_retention.py: ########## @@ -0,0 +1,278 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Celery beat task: purge soft-deleted entities past the retention window. + +The deletion-domain analog of ``version_history.prune_old_versions``: where +that ages out version rows while keeping the live entity, this removes +entities that are already soft-deleted. For each +``SoftDeleteMixin`` model it selects rows whose ``deleted_at`` is older than +the per-workspace window and runs the shared cascade per entity, in bounded +id-ordered batches. Convergent, not strictly idempotent: a re-run with the +same clock and data removes nothing, but rows that have since crossed the +cutoff are purged on a later run. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from datetime import datetime, timedelta +from typing import Any, cast + +import sqlalchemy as sa +from flask import current_app + +from superset import db +from superset.commands.deletion_retention import audit +from superset.commands.deletion_retention.purge_cascade import ( + cascade_hard_delete, + CascadeResult, + dashboard_slice_count, + entity_uuid, + suppress_purge_association_versions, +) +from superset.commands.deletion_retention.window import resolve_retention_window +from superset.extensions import celery_app, feature_flag_manager, stats_logger_manager +from superset.models.helpers import ( + skip_visibility_filter, + SoftDeleteMixin, +) + +logger: logging.Logger = logging.getLogger(__name__) + +_METRIC_PREFIX: str = "deletion_retention" +# Batch window for the eligible-id scan (SELECT ... LIMIT): bounds how many +# entities one iteration holds eligible before purging them one at a time. +_BATCH: int = 500 + + +def _soft_delete_models() -> list[type[SoftDeleteMixin]]: + """The registered ``SoftDeleteMixin`` subclasses (dashboards, charts, + datasets), in a stable order.""" + return list(SoftDeleteMixin._registered_subclasses) # noqa: SLF001 + + +def _model_table(model: type[SoftDeleteMixin]) -> sa.Table: + """Return SQLAlchemy table metadata for a registered soft-delete model.""" + return cast(sa.Table, cast(Any, model).__table__) + + +def _model_table_name(model: type[SoftDeleteMixin]) -> str: + """Return the table name for a registered soft-delete model.""" + return str(cast(Any, model).__tablename__) + + +def _iter_eligible_ids( + model: type[SoftDeleteMixin], cutoff: datetime, batch: int +) -> Iterator[list[int]]: + """Yield id-ordered batches of eligible row ids — ``deleted_at IS NOT NULL + AND deleted_at < cutoff`` — querying with the visibility-filter bypass so + soft-deleted rows are visible. Windowed by an ``id`` watermark so memory + and lock-hold stay bounded on a large first run.""" + table = _model_table(model) + after_id = 0 + while True: + with skip_visibility_filter(db.session, model): + ids = [ + row[0] + for row in db.session.execute( + sa.select(table.c.id) + .where(table.c.deleted_at.is_not(None)) + .where(table.c.deleted_at < cutoff) + .where(table.c.id > after_id) + .order_by(table.c.id) + .limit(batch) + ) + ] + if not ids: + return + yield ids + if len(ids) < batch: + return + after_id = ids[-1] + + +def _purge_impl(window_days: int, dry_run: bool) -> dict[str, Any]: + """Run one purge pass across all soft-delete models.""" + if window_days <= 0: + logger.info("deletion_retention: window is 0 (disabled); skipping") + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped") + return {"skipped": 1} + + # Same clock as SoftDeleteMixin.soft_delete(): deleted_at is stamped + # with naive-local datetime.now() (mirroring changed_on per the + # PR #33693 UTC revert), so the cutoff must be naive-local too — a + # UTC-derived cutoff would shift the retention window by the server's + # timezone offset, purging early west of UTC. If deleted_at ever moves + # to UTC-aware, this must move with it. + cutoff = datetime.now() - timedelta(days=window_days) + audit.reconcile_pending() + purged: dict[str, int] = {} Review Comment: **Suggestion:** Dry-run execution still calls `audit.reconcile_pending()`, which performs durable status updates on stale pending audit records. This means an operational dry-run can confirm or fail audit attempts even though it is documented and expected to have no side effects. Skip reconciliation when `dry_run` is enabled, or make reconciliation explicitly non-mutating in dry-run mode. [logic error] <details> <summary><b>Severity Level:</b> Major ⚠️</summary> ```mdx - ⚠️ Dry-run executions alter purge audit statuses. - ⚠️ Pending-attempt reconciliation loses dry-run side-effect isolation. - ⚠️ Operators may misinterpret audit outcomes during rollout. ``` </details> [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=414ea123be7146d1a1e127d74f917f0d&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=414ea123be7146d1a1e127d74f917f0d&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) *(Use Cmd/Ctrl + Click for best experience)* <details> <summary><b>Prompt for AI Agent 🤖 </b></summary> ```mdx This is a comment left during a code review. **Path:** superset/tasks/deletion_retention.py **Line:** 123:124 **Comment:** *Logic Error: Dry-run execution still calls `audit.reconcile_pending()`, which performs durable status updates on stale pending audit records. This means an operational dry-run can confirm or fail audit attempts even though it is documented and expected to have no side effects. Skip reconciliation when `dry_run` is enabled, or make reconciliation explicitly non-mutating in dry-run mode. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=47333ffba33da8ca9157d9e90d05ce29825fa711d9820e00d877505582ed676f&reaction=like'>👍</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=47333ffba33da8ca9157d9e90d05ce29825fa711d9820e00d877505582ed676f&reaction=dislike'>👎</a> ########## superset/tasks/deletion_retention.py: ########## @@ -0,0 +1,278 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Celery beat task: purge soft-deleted entities past the retention window. + +The deletion-domain analog of ``version_history.prune_old_versions``: where +that ages out version rows while keeping the live entity, this removes +entities that are already soft-deleted. For each +``SoftDeleteMixin`` model it selects rows whose ``deleted_at`` is older than +the per-workspace window and runs the shared cascade per entity, in bounded +id-ordered batches. Convergent, not strictly idempotent: a re-run with the +same clock and data removes nothing, but rows that have since crossed the +cutoff are purged on a later run. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from datetime import datetime, timedelta +from typing import Any, cast + +import sqlalchemy as sa +from flask import current_app + +from superset import db +from superset.commands.deletion_retention import audit +from superset.commands.deletion_retention.purge_cascade import ( + cascade_hard_delete, + CascadeResult, + dashboard_slice_count, + entity_uuid, + suppress_purge_association_versions, +) +from superset.commands.deletion_retention.window import resolve_retention_window +from superset.extensions import celery_app, feature_flag_manager, stats_logger_manager +from superset.models.helpers import ( + skip_visibility_filter, + SoftDeleteMixin, +) + +logger: logging.Logger = logging.getLogger(__name__) + +_METRIC_PREFIX: str = "deletion_retention" +# Batch window for the eligible-id scan (SELECT ... LIMIT): bounds how many +# entities one iteration holds eligible before purging them one at a time. +_BATCH: int = 500 + + +def _soft_delete_models() -> list[type[SoftDeleteMixin]]: + """The registered ``SoftDeleteMixin`` subclasses (dashboards, charts, + datasets), in a stable order.""" + return list(SoftDeleteMixin._registered_subclasses) # noqa: SLF001 + + +def _model_table(model: type[SoftDeleteMixin]) -> sa.Table: + """Return SQLAlchemy table metadata for a registered soft-delete model.""" + return cast(sa.Table, cast(Any, model).__table__) + + +def _model_table_name(model: type[SoftDeleteMixin]) -> str: + """Return the table name for a registered soft-delete model.""" + return str(cast(Any, model).__tablename__) + + +def _iter_eligible_ids( + model: type[SoftDeleteMixin], cutoff: datetime, batch: int +) -> Iterator[list[int]]: + """Yield id-ordered batches of eligible row ids — ``deleted_at IS NOT NULL + AND deleted_at < cutoff`` — querying with the visibility-filter bypass so + soft-deleted rows are visible. Windowed by an ``id`` watermark so memory + and lock-hold stay bounded on a large first run.""" + table = _model_table(model) + after_id = 0 + while True: + with skip_visibility_filter(db.session, model): + ids = [ + row[0] + for row in db.session.execute( + sa.select(table.c.id) + .where(table.c.deleted_at.is_not(None)) + .where(table.c.deleted_at < cutoff) + .where(table.c.id > after_id) + .order_by(table.c.id) + .limit(batch) + ) + ] + if not ids: + return + yield ids + if len(ids) < batch: + return + after_id = ids[-1] + + +def _purge_impl(window_days: int, dry_run: bool) -> dict[str, Any]: + """Run one purge pass across all soft-delete models.""" + if window_days <= 0: + logger.info("deletion_retention: window is 0 (disabled); skipping") + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped") + return {"skipped": 1} + + # Same clock as SoftDeleteMixin.soft_delete(): deleted_at is stamped + # with naive-local datetime.now() (mirroring changed_on per the + # PR #33693 UTC revert), so the cutoff must be naive-local too — a + # UTC-derived cutoff would shift the retention window by the server's + # timezone offset, purging early west of UTC. If deleted_at ever moves + # to UTC-aware, this must move with it. + cutoff = datetime.now() - timedelta(days=window_days) + audit.reconcile_pending() + purged: dict[str, int] = {} + would_purge: dict[str, int] = {} + failures = 0 + blocked = 0 + + for model in _soft_delete_models(): + entity_type = _model_table_name(model) + purged_n, would_n, failed_n, blocked_n = _purge_model(model, cutoff, dry_run) + if would_n: + would_purge[entity_type] = would_n + if purged_n: + purged[entity_type] = purged_n + failures += failed_n + blocked += blocked_n + + if dry_run: + for entity_type, count in would_purge.items(): + stats_logger_manager.instance.gauge( + f"{_METRIC_PREFIX}.would_purge.{entity_type}", count + ) + logger.info("deletion_retention: DRY RUN would_purge=%s", would_purge) + return {"dry_run": 1, "would_purge": would_purge} + + for entity_type, count in purged.items(): + stats_logger_manager.instance.gauge( + f"{_METRIC_PREFIX}.purged.{entity_type}", count + ) + if failures: + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.cascade_failures") + if blocked: + stats_logger_manager.instance.gauge( + f"{_METRIC_PREFIX}.blocked_by_reference", blocked + ) + stats = { + "purged": purged, + "cascade_failures": failures, + "blocked_by_reference": blocked, + } + logger.info("deletion_retention: %s", stats) + return stats + + +def _purge_model( + model: type[SoftDeleteMixin], cutoff: datetime, dry_run: bool +) -> tuple[int, int, int, int]: + """Process one model's eligible rows. Returns ``(purged, would_purge, + failures, blocked)``. A single entity's blocked/failed cascade never aborts + the batch.""" + entity_type = _model_table_name(model) + purged = would = failures = blocked = 0 + for id_batch in _iter_eligible_ids(model, cutoff, _BATCH): + if dry_run: + would += len(id_batch) + continue + for entity_id in id_batch: + try: + result = _purge_one(model, entity_id, cutoff) + if result is not None and result.purged: + purged += 1 + elif result is not None and result.blocked_reason is not None: + blocked += 1 + except Exception: # pylint: disable=broad-except + db.session.rollback() # pylint: disable=consider-using-transaction + failures += 1 + logger.exception( + "deletion_retention: cascade failed for %s id=%s", + entity_type, + entity_id, + ) + return purged, would, failures, blocked + + +def _purge_one( + model: type[SoftDeleteMixin], entity_id: int, cutoff: datetime +) -> CascadeResult | None: + """Purge a single entity in its own transaction with a write-ahead audit.""" + with skip_visibility_filter(db.session, model): + entity = db.session.get(model, entity_id) + if entity is None: + return None + entity_uuid_value = entity_uuid(entity) + removed_dashboard_slices = dashboard_slice_count(db.session, entity) + # The audit row commits on a separate connection. End the read transaction + # before that write so SQLite can later promote this session to a writer. + # Re-resolving below also ensures the cascade acts on post-audit state. + db.session.rollback() # pylint: disable=consider-using-transaction + record_id = audit.write_ahead( + trigger=audit.TRIGGER_RETENTION, + actor=audit.ACTOR_SYSTEM, + entity_type=_model_table_name(model), + entity_uuid=entity_uuid_value, + removed_dashboard_slices=removed_dashboard_slices, Review Comment: **Suggestion:** The UUID and relationship snapshot are captured before the audit write, but the entity is resolved again afterward and the cascade does not verify that the reloaded row has the same UUID. If the original row is deleted and its integer ID is reused by another eligible entity during this gap, the task can purge the replacement while recording the original UUID in the audit row. Revalidate the UUID (and ideally the original identity) immediately before cascading, or include it in the conditional claim. [race condition] <details> <summary><b>Severity Level:</b> Critical 🚨</summary> ```mdx - ❌ Concurrent purge can delete the wrong entity. - ❌ Audit UUID can identify a different entity. - ⚠️ Integer-ID reuse creates destructive race conditions. ``` </details> [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=8c888eaaf759448b977d72f5bec90e29&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=8c888eaaf759448b977d72f5bec90e29&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) *(Use Cmd/Ctrl + Click for best experience)* <details> <summary><b>Prompt for AI Agent 🤖 </b></summary> ```mdx This is a comment left during a code review. **Path:** superset/tasks/deletion_retention.py **Line:** 205:215 **Comment:** *Race Condition: The UUID and relationship snapshot are captured before the audit write, but the entity is resolved again afterward and the cascade does not verify that the reloaded row has the same UUID. If the original row is deleted and its integer ID is reused by another eligible entity during this gap, the task can purge the replacement while recording the original UUID in the audit row. Revalidate the UUID (and ideally the original identity) immediately before cascading, or include it in the conditional claim. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=7f32dd8ded45a77026a40a9a4ac361c78d1cd09317febf0cf51ec55f4a5c6f10&reaction=like'>👍</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=7f32dd8ded45a77026a40a9a4ac361c78d1cd09317febf0cf51ec55f4a5c6f10&reaction=dislike'>👎</a> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
