You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by vi...@apache.org on 2026/06/02 19:36:41 UTC
(superset) branch master updated: chore(key_value): prune expired entries from the key-value store (#40663)
This is an automated email from the ASF dual-hosted git repository.
villebro 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 8508af3201c chore(key_value): prune expired entries from the key-value store (#40663)
8508af3201c is described below
commit 8508af3201c4eafadc77f4368ceefb53ef90b60f
Author: Evan Rusackas <ev...@preset.io>
AuthorDate: Tue Jun 2 12:36:32 2026 -0700
chore(key_value): prune expired entries from the key-value store (#40663)
Co-authored-by: Claude Code <no...@anthropic.com>
Co-authored-by: Ville Brofeldt <vi...@gmail.com>
---
SECURITY.md | 2 +-
superset/config.py | 7 ++
superset/daos/key_value.py | 32 ++++++
superset/key_value/commands/__init__.py | 16 +++
superset/key_value/commands/prune.py | 143 ++++++++++++++++++++++++++
superset/tasks/scheduler.py | 16 +++
tests/unit_tests/key_value/prune_test.py | 168 +++++++++++++++++++++++++++++++
7 files changed, 383 insertions(+), 1 deletion(-)
diff --git a/SECURITY.md b/SECURITY.md
index e8ecfe56aec..a038c9a8d88 100644
--- a/SECURITY.md
+++ b/SECURITY.md
@@ -109,7 +109,7 @@ If yes, it is in scope. If no, it is out of scope. The lists below apply that te
- Any action an Admin role can perform through documented configuration, API, or UI. The Admin role is a trusted operational principal by policy. Per MITRE CNA Operational Rules 4.1, a qualifying vulnerability must violate a security policy; behavior within a documented trust boundary does not.
- Deployment or operator decisions: the values of secrets and tokens, whether internal networks are reachable from the server, which database connectors or cache backends are enabled, which feature flags are set, where notifications are delivered, and which third-party plugins are loaded.
- Compromise, modification, or malicious control of trusted backend infrastructure. Apache Superset assumes the integrity of its metastore, cache backends (for example Redis or Memcached), message brokers, secret stores, and other operator-managed infrastructure. Findings that require an attacker to read from, write to, or otherwise tamper with these systems, including injecting malicious state, serialized objects, cache entries, task metadata, configuration, or database records, are pos [...]
-- Code paths whose intended purpose is example data, demos, fixtures, local development, or documentation, rather than the production runtime.
+- The continued presence of expired key-value or metastore-cache entries that have not yet been deleted from the metadata database. Such entries are excluded from reads once expired, are purged opportunistically on write, and are removed in bulk by the scheduled `prune_key_value` maintenance task; their lingering until purged is an eventual-cleanup property, not a security boundary, and does not constitute a vulnerability.
- How a downstream application (spreadsheet program, email client, browser handling user-downloaded files) interprets output Apache Superset produced for it.
- Findings without a reproducible proof of concept against a supported release. The burden of demonstrating exploitability rests with the reporter; findings closed for lack of a proof of concept may be refiled if one is later produced.
- Brute force, rate limiting, denial of service, or resource exhaustion that does not bypass a documented control.
diff --git a/superset/config.py b/superset/config.py
index cd64e54bcc8..e2f48711a77 100644
--- a/superset/config.py
+++ b/superset/config.py
@@ -1422,6 +1422,13 @@ class CeleryConfig: # pylint: disable=too-few-public-methods
# "schedule": crontab(minute=0, hour=0),
# "kwargs": {"retention_period_days": 90, "max_rows_per_run": 10000},
# },
+ # Uncomment to enable pruning of expired entries from the key-value store
+ # (for example, rows left behind by the metastore cache backend)
+ # "prune_key_value": {
+ # "task": "prune_key_value",
+ # "schedule": crontab(minute=0, hour=0),
+ # "kwargs": {"max_rows_per_run": 10000},
+ # },
# Uncomment to enable Slack channel cache warm-up
# "slack.cache_channels": {
# "task": "slack.cache_channels",
diff --git a/superset/daos/key_value.py b/superset/daos/key_value.py
index f15293abcab..e3e63d02cf2 100644
--- a/superset/daos/key_value.py
+++ b/superset/daos/key_value.py
@@ -69,6 +69,14 @@ class KeyValueDAO(BaseDAO[KeyValueEntry]):
@staticmethod
def delete_expired_entries(resource: KeyValueResource) -> None:
+ """
+ Bulk-delete every expired entry for the given resource.
+
+ Callers that pass an explicit ``key`` to ``create_entry`` along with an
+ ``expires_on`` must invoke this once before their ``create_entry`` call(s);
+ see :meth:`create_entry` for the rationale. Expired entries are also removed
+ in bulk across all resources by the scheduled ``prune_key_value`` task.
+ """
(
db.session.query(KeyValueEntry)
.filter(
@@ -88,6 +96,20 @@ class KeyValueDAO(BaseDAO[KeyValueEntry]):
key: Key | None = None,
expires_on: datetime | None = None,
) -> KeyValueEntry:
+ """
+ Create a new entry in the key-value store.
+
+ .. note::
+ This method intentionally does **not** purge expired entries. Callers
+ that pass an explicit ``key`` along with an ``expires_on`` must call
+ :meth:`delete_expired_entries` for the same ``resource`` once before
+ creating their entries. Purging is deliberately hoisted out of this
+ method so that a transaction creating many entries pays the cleanup cost
+ only once up front rather than on every insert. An expired entry still
+ occupying the same ``key`` would otherwise cause this insert to fail the
+ unique constraint. (Entries created without an explicit ``key`` get an
+ auto-generated id and cannot collide, so they need no prior purge.)
+ """
try:
encoded_value = codec.encode(value)
except Exception as ex:
@@ -118,6 +140,12 @@ class KeyValueDAO(BaseDAO[KeyValueEntry]):
key: Key,
expires_on: datetime | None = None,
) -> KeyValueEntry:
+ """
+ Update an existing entry or create it if it does not exist.
+
+ Because this overwrites any existing entry for the key (expired or not), it
+ does not require a prior :meth:`delete_expired_entries` call.
+ """
if entry := KeyValueDAO.get_entry(resource, key):
entry.value = codec.encode(value)
entry.expires_on = expires_on
@@ -135,6 +163,10 @@ class KeyValueDAO(BaseDAO[KeyValueEntry]):
key: Key,
expires_on: datetime | None = None,
) -> KeyValueEntry:
+ """
+ Update an existing entry, raising ``KeyValueUpdateFailedError`` if no entry
+ exists for the given key.
+ """
if entry := KeyValueDAO.get_entry(resource, key):
entry.value = codec.encode(value)
entry.expires_on = expires_on
diff --git a/superset/key_value/commands/__init__.py b/superset/key_value/commands/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++ b/superset/key_value/commands/__init__.py
@@ -0,0 +1,16 @@
+# 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.
diff --git a/superset/key_value/commands/prune.py b/superset/key_value/commands/prune.py
new file mode 100644
index 00000000000..6bcf93bd6cb
--- /dev/null
+++ b/superset/key_value/commands/prune.py
@@ -0,0 +1,143 @@
+# 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.
+from __future__ import annotations
+
+import logging
+import time
+from datetime import datetime
+
+import sqlalchemy as sa
+
+from superset import db
+from superset.commands.base import BaseCommand
+from superset.key_value.models import KeyValueEntry
+
+logger = logging.getLogger(__name__)
+
+
+# pylint: disable=consider-using-transaction
+class KeyValuePruneCommand(BaseCommand):
+ """
+ Command to prune the key-value store by deleting entries whose expiry has
+ already passed.
+
+ The metastore-backed key-value store keeps an `expires_on` timestamp for
+ entries written with a timeout (for example, the metastore cache backend).
+ Unlike cache backends that evict on read, the metastore does not remove rows
+ on its own, so expired entries remain in the table until something deletes
+ them. This command performs that housekeeping by deleting every entry whose
+ `expires_on` is in the past, freeing up space in the table.
+
+ Attributes:
+ max_rows_per_run (int | None): The maximum number of rows to delete in a
+ single run. If provided and greater than
+ zero, rows are selected deterministically
+ from the oldest first by id up to this
+ limit in this execution.
+ """
+
+ def __init__(self, max_rows_per_run: int | None = None):
+ """
+ :param max_rows_per_run: The maximum number of rows to delete in a single run.
+ If provided and greater than zero, rows are selected deterministically from
+ the oldest first by id up to this limit in this execution.
+ """
+ self.max_rows_per_run = max_rows_per_run
+
+ def run(self) -> None:
+ """
+ Executes the prune command
+ """
+ batch_size = 999 # SQLite has a IN clause limit of 999
+ total_deleted = 0
+ start_time = time.time()
+
+ # Capture a single cutoff timestamp and reuse it for both the selection
+ # and the delete. Reusing the same value (rather than calling
+ # datetime.now() again at delete time) keeps the delete predicate
+ # consistent with the selection and re-checks expiry on delete, so an
+ # entry refreshed (expires_on moved into the future) between selection
+ # and deletion is not removed.
+ cutoff = datetime.now()
+
+ # Select all IDs whose expiry has already passed. Entries without an
+ # expiry (expires_on IS NULL) never expire and are left untouched. The
+ # `<=` comparison matches KeyValueEntry.is_expired() and
+ # KeyValueDAO.delete_expired_entries() so all expiry checks agree.
+ select_stmt = sa.select(KeyValueEntry.id).where(
+ KeyValueEntry.expires_on.isnot(None),
+ KeyValueEntry.expires_on <= cutoff,
+ )
+
+ # Optionally limited by max_rows_per_run
+ # order by oldest first for deterministic deletion
+ if self.max_rows_per_run is not None and self.max_rows_per_run > 0:
+ select_stmt = select_stmt.order_by(KeyValueEntry.id.asc()).limit(
+ self.max_rows_per_run
+ )
+
+ ids_to_delete = db.session.execute(select_stmt).scalars().all()
+
+ total_rows = len(ids_to_delete)
+
+ logger.info("Total rows to be deleted: %s", f"{total_rows:,}")
+
+ next_logging_threshold = 1
+
+ # Iterate over the IDs in batches
+ for i in range(0, total_rows, batch_size):
+ batch_ids = ids_to_delete[i : i + batch_size]
+
+ # Delete the selected batch using IN clause. The expiry predicate is
+ # re-applied here (against the same cutoff captured before selection)
+ # so an entry refreshed between selection and deletion is left intact.
+ result = db.session.execute(
+ sa.delete(KeyValueEntry).where(
+ KeyValueEntry.id.in_(batch_ids),
+ KeyValueEntry.expires_on.isnot(None),
+ KeyValueEntry.expires_on <= cutoff,
+ )
+ )
+
+ # Update the total number of deleted records
+ total_deleted += result.rowcount
+
+ # Explicitly commit the transaction so that, if an error occurs, the
+ # records deleted so far are committed
+ db.session.commit()
+
+ # Log the number of deleted records every 1% increase in progress
+ percentage_complete = (total_deleted / total_rows) * 100
+ if percentage_complete >= next_logging_threshold:
+ logger.info(
+ "Deleted %s expired rows from the key-value store (%d%% complete)",
+ f"{total_deleted:,}",
+ percentage_complete,
+ )
+ next_logging_threshold += 1
+
+ elapsed_time = time.time() - start_time
+ minutes, seconds = divmod(elapsed_time, 60)
+ formatted_time = f"{int(minutes):02}:{int(seconds):02}"
+ logger.info(
+ "Pruning complete: %s expired rows deleted in %s",
+ f"{total_deleted:,}",
+ formatted_time,
+ )
+
+ def validate(self) -> None:
+ pass
diff --git a/superset/tasks/scheduler.py b/superset/tasks/scheduler.py
index 5b38048adf1..a37c049a06a 100644
--- a/superset/tasks/scheduler.py
+++ b/superset/tasks/scheduler.py
@@ -38,6 +38,7 @@ from superset.commands.tasks.prune import TaskPruneCommand
from superset.daos.report import ReportScheduleDAO
from superset.daos.tasks import TaskDAO
from superset.extensions import celery_app
+from superset.key_value.commands.prune import KeyValuePruneCommand
from superset.stats_logger import BaseStatsLogger
from superset.tasks.ambient_context import use_context
from superset.tasks.constants import ABORT_STATES, TERMINAL_STATES
@@ -236,6 +237,21 @@ def prune_tasks(
logger.exception("An error occurred while pruning async tasks: %s", ex)
+@celery_app.task(name="prune_key_value", bind=True)
+def prune_key_value(
+ self: Task,
+ max_rows_per_run: int | None = None,
+ **kwargs: Any,
+) -> None:
+ stats_logger: BaseStatsLogger = current_app.config["STATS_LOGGER"]
+ stats_logger.incr("prune_key_value")
+
+ try:
+ KeyValuePruneCommand(max_rows_per_run).run()
+ except CommandException as ex:
+ logger.exception("An error occurred while pruning the key-value store: %s", ex)
+
+
@celery_app.task(name="tasks.execute", bind=True)
def execute_task( # noqa: C901
self: Any, # Celery task instance
diff --git a/tests/unit_tests/key_value/prune_test.py b/tests/unit_tests/key_value/prune_test.py
new file mode 100644
index 00000000000..f69b56bf051
--- /dev/null
+++ b/tests/unit_tests/key_value/prune_test.py
@@ -0,0 +1,168 @@
+# 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.
+# pylint: disable=import-outside-toplevel, unused-argument
+from __future__ import annotations
+
+from collections.abc import Generator
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+from flask.ctx import AppContext
+
+from superset.extensions import db
+from superset.key_value.types import JsonKeyValueCodec, KeyValueResource
+
+if TYPE_CHECKING:
+ from superset.key_value.models import KeyValueEntry
+
+RESOURCE = KeyValueResource.METASTORE_CACHE
+CODEC = JsonKeyValueCodec()
+VALUE = {"foo": "bar"}
+
+
+@pytest.fixture
+def clean_key_value_store(app_context: AppContext) -> Generator[None, None, None]:
+ # The prune command commits, so a plain session rollback cannot undo its
+ # effects. Explicitly empty the table around each test for isolation.
+ from superset.key_value.models import KeyValueEntry
+
+ db.session.query(KeyValueEntry).delete()
+ db.session.commit() # pylint: disable=consider-using-transaction
+ yield
+ db.session.query(KeyValueEntry).delete()
+ db.session.commit() # pylint: disable=consider-using-transaction
+
+
+def _add_entry(expires_on: datetime | None) -> KeyValueEntry:
+ from superset.key_value.models import KeyValueEntry
+
+ entry = KeyValueEntry(
+ resource=RESOURCE,
+ value=CODEC.encode(VALUE),
+ expires_on=expires_on,
+ )
+ db.session.add(entry)
+ db.session.flush()
+ return entry
+
+
+def test_prune_deletes_expired_entries(
+ clean_key_value_store: None,
+) -> None:
+ from superset.key_value.commands.prune import KeyValuePruneCommand
+ from superset.key_value.models import KeyValueEntry
+
+ expired_a = _add_entry(datetime.now() - timedelta(days=1))
+ expired_b = _add_entry(datetime.now() - timedelta(seconds=5))
+ expired_ids = {expired_a.id, expired_b.id}
+
+ KeyValuePruneCommand().run()
+
+ remaining_ids = {row.id for row in db.session.query(KeyValueEntry.id).all()}
+ assert expired_ids.isdisjoint(remaining_ids)
+ assert db.session.query(KeyValueEntry).count() == 0
+
+
+def test_prune_retains_non_expired_and_no_expiry_entries(
+ clean_key_value_store: None,
+) -> None:
+ from superset.key_value.commands.prune import KeyValuePruneCommand
+ from superset.key_value.models import KeyValueEntry
+
+ future = _add_entry(datetime.now() + timedelta(days=1))
+ no_expiry = _add_entry(None)
+ expired = _add_entry(datetime.now() - timedelta(days=1))
+
+ KeyValuePruneCommand().run()
+
+ remaining_ids = {row.id for row in db.session.query(KeyValueEntry.id).all()}
+ assert future.id in remaining_ids
+ assert no_expiry.id in remaining_ids
+ assert expired.id not in remaining_ids
+ assert db.session.query(KeyValueEntry).count() == 2
+
+
+def test_prune_empty_store_is_noop(
+ clean_key_value_store: None,
+) -> None:
+ from superset.key_value.commands.prune import KeyValuePruneCommand
+ from superset.key_value.models import KeyValueEntry
+
+ assert db.session.query(KeyValueEntry).count() == 0
+
+ # Should not raise and should leave the store empty
+ KeyValuePruneCommand().run()
+
+ assert db.session.query(KeyValueEntry).count() == 0
+
+
+def test_prune_skips_entry_refreshed_after_selection(
+ clean_key_value_store: None,
+) -> None:
+ from superset.key_value.commands.prune import KeyValuePruneCommand
+ from superset.key_value.models import KeyValueEntry
+
+ # Simulate the TOCTOU window: an entry is expired when prune selects it but
+ # is refreshed (expires_on moved into the future) before the delete runs.
+ # The delete re-checks expiry against the cutoff captured at selection time,
+ # so the refreshed entry must survive. We inject the refresh right after the
+ # command captures its cutoff by patching datetime.now used in the command.
+ expired = _add_entry(datetime.now() - timedelta(days=1))
+ db.session.commit() # pylint: disable=consider-using-transaction
+
+ import superset.key_value.commands.prune as prune_module
+
+ real_now = datetime.now
+ state = {"refreshed": False}
+
+ class _PatchedDatetime:
+ @staticmethod
+ def now() -> datetime:
+ # The command calls now() once to capture the cutoff. After that
+ # call, refresh the entry so the subsequent delete sees a future
+ # expires_on but still deletes against the original cutoff.
+ current = real_now()
+ if not state["refreshed"]:
+ state["refreshed"] = True
+ db.session.query(KeyValueEntry).filter(
+ KeyValueEntry.id == expired.id
+ ).update({KeyValueEntry.expires_on: real_now() + timedelta(days=1)})
+ db.session.commit() # pylint: disable=consider-using-transaction
+ return current
+
+ with patch.object(prune_module, "datetime", _PatchedDatetime):
+ KeyValuePruneCommand().run()
+
+ remaining_ids = {row.id for row in db.session.query(KeyValueEntry.id).all()}
+ assert expired.id in remaining_ids
+
+
+def test_prune_respects_max_rows_per_run(
+ clean_key_value_store: None,
+) -> None:
+ from superset.key_value.commands.prune import KeyValuePruneCommand
+ from superset.key_value.models import KeyValueEntry
+
+ for _ in range(3):
+ _add_entry(datetime.now() - timedelta(days=1))
+
+ KeyValuePruneCommand(max_rows_per_run=2).run()
+
+ # Only the two oldest expired rows are removed in a single capped run
+ assert db.session.query(KeyValueEntry).count() == 1