feat(soft-delete): warn at startup when the purge beat entry is missing

superset/initialization/__init__.py

@@ -881,17 +881,31 @@ def init_versioning(self) -> None:
         # retention work doesn't add latency to user saves.
 
     _RETENTION_TASK_NAME: str = "version_history.prune_old_versions"
+    _PURGE_TASK_NAME: str = "deletion_retention.purge_soft_deleted"
 
     def _warn_if_retention_beat_missing(self) -> None:
-        """WARN at startup when the resolved Celery beat schedule has no
-        ``version_history.prune_old_versions`` entry.
+        """WARN at startup when the resolved Celery beat schedule is
+        missing a time-based retention task:
+
+        * ``version_history.prune_old_versions`` — checked always, since
+          shadow rows written by prior deploys keep ageing even when
+          capture is off;
+        * ``deletion_retention.purge_soft_deleted`` — checked only when
+          the ``SOFT_DELETE`` feature flag resolves on at the config
+          level, because the purge task itself no-ops while the flag is
+          off, so a missing entry is only actionable once soft delete is
+          live. (Flags supplied dynamically via ``GET_FEATURE_FLAGS_FUNC``
+          are not consulted here; this startup check reads configuration
+          only.)
 
         Operators who redefine ``CeleryConfig`` in ``superset_config.py``
         — instead of subclassing or merging the default — silently lose
-        the retention task. Capture continues writing rows; the prune
-        never runs; disk grows until paged. The default config carries
-        the entry; this check makes the misconfiguration visible in the
-        deploy log before disk pressure makes it visible at 03:00.
+        these tasks. Capture continues writing rows; the prune
+        never runs; disk grows until paged. Archived objects likewise
+        accumulate forever instead of purging after the retention window.
+        The default config carries both entries; this check makes the
+        misconfiguration visible in the deploy log before disk pressure
+        makes it visible at 03:00.
 
         Handles four shapes of ``CELERY_CONFIG``:
         * ``None`` — Celery deliberately disabled, no retention either
@@ -934,6 +948,24 @@ def _warn_if_retention_beat_missing(self) -> None:
                 "default CeleryConfig or add the entry to your override.",
                 self._RETENTION_TASK_NAME,
             )
+        # Config-level flag resolution: the deployment's explicit
+        # FEATURE_FLAGS overrides the shipped defaults, mirroring how
+        # FeatureFlagManager.init_app seeds its dict before dynamic
+        # overrides.
+        feature_flags = {
+            **self.config.get("DEFAULT_FEATURE_FLAGS", {}),
+            **self.config.get("FEATURE_FLAGS", {}),
+        }
+        if feature_flags.get("SOFT_DELETE") and (
+            not beat_schedule or self._PURGE_TASK_NAME not in registered_tasks
+        ):
+            logger.warning(
+                "soft-delete: CELERY_CONFIG.beat_schedule is missing the "
+                "%r entry — archived objects will never be purged and "
+                "will accumulate indefinitely. Either inherit from the "
+                "default CeleryConfig or add the entry to your override.",
+                self._PURGE_TASK_NAME,
+            )
 
     def init_app_in_ctx(self) -> None:
         """

tests/unit_tests/initialization_test.py

@@ -724,3 +724,152 @@ def test_string_celery_config_reference_does_not_warn(
         initializer._warn_if_retention_beat_missing()
 
         mock_logger.warning.assert_not_called()
+
+    @patch("superset.initialization.logger")
+    def test_warn_when_soft_delete_on_and_purge_entry_missing(
+        self, mock_logger: MagicMock
+    ) -> None:
+        """With ``SOFT_DELETE`` resolving on at the config level and the
+        beat schedule carrying the version-history entry but not the
+        purge entry, a WARNING naming
+        ``deletion_retention.purge_soft_deleted`` fires — otherwise a
+        hand-rolled ``CeleryConfig`` silently accumulates archived
+        objects forever."""
+
+        class _NoPurgeCeleryConfig:
+            beat_schedule: dict[str, dict[str, str]] = {
+                "version_history.prune_old_versions": {
+                    "task": "version_history.prune_old_versions",
+                },
+            }
+
+        initializer = self._initializer(
+            {
+                "CELERY_CONFIG": _NoPurgeCeleryConfig,
+                "DEFAULT_FEATURE_FLAGS": {"SOFT_DELETE": True},
+                "FEATURE_FLAGS": {},
+            }
+        )
+        initializer._warn_if_retention_beat_missing()
+
+        assert any(
+            "deletion_retention.purge_soft_deleted" in str(call)
+            for call in mock_logger.warning.call_args_list
+        ), (
+            "Expected a WARNING naming the missing purge entry; "
+            f"got {mock_logger.warning.call_args_list}"
+        )
+
+    @patch("superset.initialization.logger")
+    def test_no_purge_warn_when_soft_delete_off(self, mock_logger: MagicMock) -> None:
+        """With ``SOFT_DELETE`` off, a missing purge entry MUST NOT warn:
+        the purge task itself no-ops while the flag is off, so the
+        warning would be noise the operator cannot act on."""
+
+        class _NoPurgeCeleryConfig:
+            beat_schedule: dict[str, dict[str, str]] = {
+                "version_history.prune_old_versions": {
+                    "task": "version_history.prune_old_versions",
+                },
+            }
+
+        initializer = self._initializer(
+            {
+                "CELERY_CONFIG": _NoPurgeCeleryConfig,
+                "DEFAULT_FEATURE_FLAGS": {"SOFT_DELETE": False},
+                "FEATURE_FLAGS": {},
+            }
+        )
+        initializer._warn_if_retention_beat_missing()
+
+        mock_logger.warning.assert_not_called()
+
+    @patch("superset.initialization.logger")
+    def test_feature_flags_override_beats_default_for_purge_warn(
+        self, mock_logger: MagicMock
+    ) -> None:
+        """A deployment's explicit ``FEATURE_FLAGS`` entry overrides the
+        shipped default, matching ``FeatureFlagManager.init_app`` merge
+        order: default off + override on must warn on a missing purge
+        entry."""
+
+        class _NoPurgeCeleryConfig:
+            beat_schedule: dict[str, dict[str, str]] = {
+                "version_history.prune_old_versions": {
+                    "task": "version_history.prune_old_versions",
+                },
+            }
+
+        initializer = self._initializer(
+            {
+                "CELERY_CONFIG": _NoPurgeCeleryConfig,
+                "DEFAULT_FEATURE_FLAGS": {"SOFT_DELETE": False},
+                "FEATURE_FLAGS": {"SOFT_DELETE": True},
+            }
+        )
+        initializer._warn_if_retention_beat_missing()
+
+        assert any(
+            "deletion_retention.purge_soft_deleted" in str(call)
+            for call in mock_logger.warning.call_args_list
+        ), (
+            "Expected the FEATURE_FLAGS override to enable the purge "
+            f"check; got {mock_logger.warning.call_args_list}"
+        )
+
+    @patch("superset.initialization.logger")
+    def test_no_warn_when_soft_delete_on_and_both_entries_present(
+        self, mock_logger: MagicMock
+    ) -> None:
+        """The default ``CeleryConfig`` shape with soft delete enabled:
+        both retention entries present, no warning at all."""
+
+        class _CompleteCeleryConfig:
+            beat_schedule: dict[str, dict[str, str]] = {
+                "version_history.prune_old_versions": {
+                    "task": "version_history.prune_old_versions",
+                },
+                "deletion_retention.purge_soft_deleted": {
+                    "task": "deletion_retention.purge_soft_deleted",
+                },
+            }
+
+        initializer = self._initializer(
+            {
+                "CELERY_CONFIG": _CompleteCeleryConfig,
+                "DEFAULT_FEATURE_FLAGS": {"SOFT_DELETE": True},
+                "FEATURE_FLAGS": {},
+            }
+        )
+        initializer._warn_if_retention_beat_missing()
+
+        mock_logger.warning.assert_not_called()
+
+    @patch("superset.initialization.logger")
+    def test_no_purge_warn_when_task_registered_under_other_key(
+        self, mock_logger: MagicMock
+    ) -> None:
+        """Parity with the version-history check: the purge task
+        registered under a non-matching schedule key is still correctly
+        scheduled and MUST NOT warn."""
+
+        class _RenamedKeysCeleryConfig:
+            beat_schedule: dict[str, dict[str, str]] = {
+                "prune_versions": {
+                    "task": "version_history.prune_old_versions",
+                },
+                "purge_archived": {
+                    "task": "deletion_retention.purge_soft_deleted",
+                },
+            }
+
+        initializer = self._initializer(
+            {
+                "CELERY_CONFIG": _RenamedKeysCeleryConfig,
+                "DEFAULT_FEATURE_FLAGS": {"SOFT_DELETE": True},
+                "FEATURE_FLAGS": {},
+            }
+        )
+        initializer._warn_if_retention_beat_missing()
+
+        mock_logger.warning.assert_not_called()