You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kalluripradeep (via GitHub)" <gi...@apache.org> on 2026/06/21 09:36:39 UTC
[PR] fix: MetastoreBackend.cleanup() ignores default_retention_days for null expires_at rows [airflow]
kalluripradeep opened a new pull request, #68793:
URL: https://github.com/apache/airflow/pull/68793
MetastoreBackend.cleanup() only deleted rows where expires_at < now(). Task workers always write state with expires_at=None (the default), so the configured [state_store] default_retention_days (default: 30 days) was silently ignored — those rows accumulated in task_state_store forever.
The model's own comment, the CLI description, and the config docs all state that null-expires_at rows are cleaned up by a "global updated_at + default_retention_days check" — but that check was never implemented.
Fix
Add a second deletion pass in cleanup() and _summary_dry_run() that targets rows where expires_at IS NULL AND updated_at < now - timedelta(days=default_retention_days). The pass is skipped when default_retention_days=0.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: MetastoreBackend.cleanup() ignores default_retention_days for null expires_at rows [airflow]
Posted by "kalluripradeep (via GitHub)" <gi...@apache.org>.
kalluripradeep commented on PR #68793:
URL: https://github.com/apache/airflow/pull/68793#issuecomment-4766076162
@amoghrajesh You're absolutely right — thank you for the correction.
I traced the SDK code and found the critical comment at line 580 of `context.py`:
```python
# expires_at is always resolved on the worker in UTC before being sent.
```
The full resolution logic in `TaskStateStoreAccessor.set()`:
```python
if retention is NEVER_EXPIRE:
expires_at = None # → NULL in DB
elif retention is not None:
expires_at = now + retention # → explicit datetime
else: # retention=None (default)
days = conf.getint("state_store", "default_retention_days")
expires_at = None if days == 0 else now + timedelta(days=days) # → datetime or NULL (if days=0)
```
So a plain `task_state_store.set("key", value)` with `default_retention_days=30` writes `expires_at = now + 30 days` to the DB — and the existing `WHERE expires_at < now()` cleanup handles that correctly after 30 days.
`expires_at=NULL` in the DB means either `NEVER_EXPIRE` was explicitly passed, or `default_retention_days=0` (disabled). The PR would incorrectly delete those rows, breaking the `NEVER_EXPIRE` contract.
The bug I reported doesn't exist. Closing this PR and the linked issue.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: MetastoreBackend.cleanup() ignores default_retention_days for null expires_at rows [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #68793:
URL: https://github.com/apache/airflow/pull/68793#discussion_r3448963282
##########
airflow-core/src/airflow/state/metastore.py:
##########
@@ -383,13 +383,24 @@ def _clear_asset_state_store(self, scope: AssetScope, *, session: Session) -> No
def cleanup(self) -> None:
"""
- Remove expired task state rows.
+ Remove expired and stale task state rows.
- ``expires_at`` is set at write time on every ``set()`` call, so cleanup is a single
- ``WHERE expires_at < now()`` pass. Rows with ``expires_at=NULL`` (default_retention_days=0)
- are never deleted. Batching is configurable via ``[state_store] state_cleanup_batch_size``.
+ Two deletion passes run in order:
+
+ 1. **Explicit expiry**: rows where ``expires_at IS NOT NULL AND expires_at < now()``.
+ These are rows where the writer set a specific TTL via ``set(expires_at=...)``.
+
+ 2. **Default retention**: rows where ``expires_at IS NULL`` and
+ ``updated_at + timedelta(days=default_retention_days) < now()``.
+ This applies the deployment-wide ``[state_store] default_retention_days`` (default 30)
+ to rows that were written without an explicit TTL — which is the common case for
+ task worker writes. When ``default_retention_days`` is 0, this pass is skipped
+ and those rows are kept indefinitely.
+
+ Batching is configurable via ``[state_store] state_cleanup_batch_size``.
Review Comment:
This docstring is too long and goes far too deep into implementation details. I woud suggest the below instead:
```
"""
Remove expired task state rows.
Rows with an explicit ``expires_at`` are deleted when expired. Rows without an
explicit expiry are deleted according to ``[state_store] default_retention_days``.
"""
```
##########
airflow-core/tests/unit/state/test_metastore.py:
##########
@@ -260,6 +260,123 @@ def test_set_expires_at_none_stores_null(
assert row is not None
assert row.expires_at is None
+ def test_cleanup_deletes_stale_rows_by_default_retention(
+ self, session: Session, backend: MetastoreBackend, dag_run: DagRun
+ ):
+ """cleanup() removes rows with expires_at=NULL that exceed default_retention_days.
+
+ This is the primary regression test for the bug where cleanup() only checked
+ ``expires_at < now()`` and completely ignored ``default_retention_days``, causing
+ rows written by the task worker (which send expires_at=None) to accumulate forever.
Review Comment:
I would suggest the below to be more concise:
`"""cleanup() removes stale NULL-expires_at rows using default_retention_days."""`
##########
airflow-core/src/airflow/state/metastore.py:
##########
@@ -409,11 +420,25 @@ def _delete_batched(where_clause) -> int:
break
return total
+ # Pass 1: explicit expires_at
deleted = _delete_batched(TaskStateStoreModel.expires_at < now)
- log.info("Deleted expired task_state_store rows", rows_deleted=deleted)
+ log.info("Deleted explicitly-expired task_state_store rows", rows_deleted=deleted)
+
+ # Pass 2: default retention for rows without an explicit expires_at
+ if retention_days > 0:
+ cutoff = now - timedelta(days=retention_days)
+ stale_deleted = _delete_batched(
+ (TaskStateStoreModel.expires_at.is_(None)) & (TaskStateStoreModel.updated_at < cutoff)
+ )
+ log.info(
+ "Deleted stale task_state_store rows by default retention",
+ rows_deleted=stale_deleted,
+ retention_days=retention_days,
+ )
Review Comment:
I am just curious why you split the cleanup across 2 separate database queries? Why not do something like this instead:
```
_delete_batched(
(TaskStateStoreModel.expires_at < now)
| (
TaskStateStoreModel.expires_at.is_(None)
& (TaskStateStoreModel.updated_at < cutoff)
)
)
```
Is this solely for better observability? In that case, would user need to know the precise breakdown of `expires_at` and 'default retention' rows that have been deleted?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: MetastoreBackend.cleanup() ignores default_retention_days for null expires_at rows [airflow]
Posted by "kalluripradeep (via GitHub)" <gi...@apache.org>.
kalluripradeep commented on code in PR #68793:
URL: https://github.com/apache/airflow/pull/68793#discussion_r3449062042
##########
airflow-core/src/airflow/state/metastore.py:
##########
@@ -383,13 +383,24 @@ def _clear_asset_state_store(self, scope: AssetScope, *, session: Session) -> No
def cleanup(self) -> None:
"""
- Remove expired task state rows.
+ Remove expired and stale task state rows.
- ``expires_at`` is set at write time on every ``set()`` call, so cleanup is a single
- ``WHERE expires_at < now()`` pass. Rows with ``expires_at=NULL`` (default_retention_days=0)
- are never deleted. Batching is configurable via ``[state_store] state_cleanup_batch_size``.
+ Two deletion passes run in order:
+
+ 1. **Explicit expiry**: rows where ``expires_at IS NOT NULL AND expires_at < now()``.
+ These are rows where the writer set a specific TTL via ``set(expires_at=...)``.
+
+ 2. **Default retention**: rows where ``expires_at IS NULL`` and
+ ``updated_at + timedelta(days=default_retention_days) < now()``.
+ This applies the deployment-wide ``[state_store] default_retention_days`` (default 30)
+ to rows that were written without an explicit TTL — which is the common case for
+ task worker writes. When ``default_retention_days`` is 0, this pass is skipped
+ and those rows are kept indefinitely.
+
+ Batching is configurable via ``[state_store] state_cleanup_batch_size``.
Review Comment:
Good call — updated. Trimmed down to just the two-sentence summary you suggested.
##########
airflow-core/src/airflow/state/metastore.py:
##########
@@ -409,11 +420,25 @@ def _delete_batched(where_clause) -> int:
break
return total
+ # Pass 1: explicit expires_at
deleted = _delete_batched(TaskStateStoreModel.expires_at < now)
- log.info("Deleted expired task_state_store rows", rows_deleted=deleted)
+ log.info("Deleted explicitly-expired task_state_store rows", rows_deleted=deleted)
+
+ # Pass 2: default retention for rows without an explicit expires_at
+ if retention_days > 0:
+ cutoff = now - timedelta(days=retention_days)
+ stale_deleted = _delete_batched(
+ (TaskStateStoreModel.expires_at.is_(None)) & (TaskStateStoreModel.updated_at < cutoff)
+ )
+ log.info(
+ "Deleted stale task_state_store rows by default retention",
+ rows_deleted=stale_deleted,
+ retention_days=retention_days,
+ )
Review Comment:
You're right, the split was only for observability and it's not worth the extra complexity. Merged both passes into a single `_delete_batched` call with an OR clause (adds the default-retention arm only when `retention_days > 0`). Single log line is sufficient.
##########
airflow-core/tests/unit/state/test_metastore.py:
##########
@@ -260,6 +260,123 @@ def test_set_expires_at_none_stores_null(
assert row is not None
assert row.expires_at is None
+ def test_cleanup_deletes_stale_rows_by_default_retention(
+ self, session: Session, backend: MetastoreBackend, dag_run: DagRun
+ ):
+ """cleanup() removes rows with expires_at=NULL that exceed default_retention_days.
+
+ This is the primary regression test for the bug where cleanup() only checked
+ ``expires_at < now()`` and completely ignored ``default_retention_days``, causing
+ rows written by the task worker (which send expires_at=None) to accumulate forever.
Review Comment:
Done — replaced with the one-liner you suggested.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: MetastoreBackend.cleanup() ignores default_retention_days for null expires_at rows [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh closed pull request #68793: fix: MetastoreBackend.cleanup() ignores default_retention_days for null expires_at rows
URL: https://github.com/apache/airflow/pull/68793
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org