This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 4a059ea539b Stop ObjectStorage journal write failures from failing
durable tasks (#75916)
4a059ea539b is described below
commit 4a059ea539bd0d6558fdb0a97d699ed0eae2feda
Author: Yash jain <[email protected]>
AuthorDate: Thu Nov 12 01:16:57 2026 +0530
Stop ObjectStorage journal write failures from failing durable tasks
(#75916)
The DurableStorageProtocol contract says a backend may skip a write without
failing the step, and the task state store backend (Airflow >= 3.3) does.
The ObjectStorage backend did not: a failed write raised out of save_step,
which could fail a task whose work had already succeeded (journal cleanup),
replace the agent's own error (DurableRun.fail), or fail a tool step after
its side effect ran.
Reads were too lenient in the other direction: any OSError or corrupt file
was silently treated as an empty journal, and the next save then rewrote the
file with only this attempt's steps, discarding the previous attempt's.
Now a failed write skips that step with a warning and leaves the in-memory
journal as it was; a journal file that exists but cannot be read is left in
place and this attempt runs without journaling; a file that is not a valid
journal is reported and replaced.
* Treat a durable journal file that is not UTF-8 text as invalid
* Reword the DurableStorage docstring for the docs spell check
* Keep only the write guard in durable journal storage
Per review, reads behave as on main again, so the unreadable-journal
state, the UnicodeDecodeError case and the docs paragraph about reads
are gone. save_step builds the new journal text first and updates the
in-memory cache only once the write succeeds, which drops _MISSING and
_restore. A journal holding valid JSON that is not an object is now
handled like one cut off mid-write instead of failing with an
AttributeError.
---
providers/common/ai/docs/durable_execution.rst | 4 +-
.../airflow/providers/common/ai/durable/storage.py | 28 ++++++---
.../tests/unit/common/ai/durable/test_storage.py | 73 ++++++++++++++++++++++
3 files changed, 95 insertions(+), 10 deletions(-)
diff --git a/providers/common/ai/docs/durable_execution.rst
b/providers/common/ai/docs/durable_execution.rst
index 1beaf3ef032..7d986a1ad5d 100644
--- a/providers/common/ai/docs/durable_execution.rst
+++ b/providers/common/ai/docs/durable_execution.rst
@@ -375,8 +375,8 @@ depending on the pydantic-ai version and the toolset, such
a result fails the ta
it comes back changed (as text, a list or a dict), which can make the steps
after that
tool call run again. Put bytes in a dict instead. A result the journal cannot
encode,
such as an object with no JSON form, fails the task with
``DurableJournalError`` and no
-retry, since it would fail the same way every time. A write the task state
store
-refuses does not fail the task: the task log names the step, and a retry runs
it again.
+retry, since it would fail the same way every time. A write the storage
refuses or
+fails does not fail the task: the task log names the step, and a retry runs it
again.
Limitations
-----------
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
b/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
index 5a41b788dcb..e566903a63d 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
@@ -92,10 +92,13 @@ class DurableStorage:
path = self._get_path()
try:
- self._cache = json.loads(path.read_text())
+ cache = json.loads(path.read_text())
+ if not isinstance(cache, dict):
+ # JSON that is not an object holds no steps, just like a
cut-off file.
+ raise ValueError("the journal file does not hold a JSON
object")
except (FileNotFoundError, OSError, json.JSONDecodeError, ValueError):
- self._cache = {}
-
+ cache = {}
+ self._cache = cache
return self._cache
def _write(self, text: str) -> None:
@@ -108,18 +111,27 @@ class DurableStorage:
"""
Store one journal step and rewrite the journal file.
- :return: ``True`` if the entry was written, ``False`` if it could not
be encoded.
+ Best-effort: the save runs after the step already succeeded, so a
failed write
+ must not fail the step. It is skipped with a warning, and the step
runs again on
+ the next retry.
+
+ :return: ``True`` if the entry was written, ``False`` if it could not
be encoded
+ or the write failed.
"""
cache = self._load_cache()
- cache[key] = entry
try:
- text = json.dumps(cache)
+ text = json.dumps({**cache, key: entry})
except (TypeError, ValueError):
# Only this entry is skipped; the rest of the journal is still
written.
- del cache[key]
log.warning("Durable: could not store step", key=key,
exc_info=True)
return False
- self._write(text)
+ try:
+ self._write(text)
+ except Exception:
+ # The step already ran, so a failed write must not fail it; it
runs again on retry.
+ log.warning("Durable: could not write the journal file", key=key,
exc_info=True)
+ return False
+ cache[key] = entry
return True
def load_step(self, key: str) -> dict[str, Any] | None:
diff --git a/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
b/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
index a9098cc1922..09ed9414cdf 100644
--- a/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
+++ b/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
@@ -20,8 +20,10 @@ import json
from unittest.mock import patch
import pytest
+from structlog.typing import FilteringBoundLogger
from airflow.providers.common.ai.durable import storage as storage_module
+from airflow.providers.common.ai.durable.journal import DurableJournal
from airflow.providers.common.ai.durable.storage import DurableStorage
from airflow.providers.common.compat.sdk import ObjectStoragePath
@@ -126,6 +128,77 @@ class TestDeleteSteps:
storage.delete_steps(["a"])
+def store_down():
+ return patch.object(DurableStorage, "_write", autospec=True,
side_effect=OSError("store unavailable"))
+
+
+class TestStoreFailures:
+ def test_a_failed_write_skips_the_step_without_raising(self, storage):
+ with store_down(), patch.object(storage_module, "log",
spec=FilteringBoundLogger) as mock_log:
+ assert not storage.save_step("k", ENTRY)
+
+ assert storage.load_step("k") is None
+ mock_log.warning.assert_called_once_with(
+ "Durable: could not write the journal file", key="k", exc_info=True
+ )
+
+ def test_a_failed_write_keeps_the_previous_value_of_a_step(self, storage):
+ storage.save_step("meta", {"high_water": 1})
+
+ with store_down():
+ assert not storage.save_step("meta", {"high_water": 2})
+
+ assert storage.load_step("meta") == {"high_water": 1}
+
+ def
test_an_entry_that_cannot_be_encoded_keeps_the_previous_value_of_a_step(self,
storage):
+ storage.save_step("meta", {"high_water": 1})
+
+ assert not storage.save_step("meta", {"high_water": object()})
+
+ assert storage.load_step("meta") == {"high_water": 1}
+
+ def test_a_skipped_step_is_not_written_by_a_later_save(self, storage,
tmp_path):
+ with store_down():
+ storage.save_step("skipped", ENTRY)
+ storage.save_step("written", ENTRY)
+
+ (journal_file,) = tmp_path.iterdir()
+ assert set(json.loads(journal_file.read_text())) == {"written"}
+
+ def test_an_outage_after_the_work_succeeded_does_not_fail_the_task(self,
storage):
+ journal = DurableJournal(storage)
+ first, second = journal.start_run(), journal.start_run()
+ first.save(0, ENTRY)
+ second.save(0, ENTRY)
+
+ with store_down():
+ journal.cleanup()
+
+ def test_an_outage_does_not_replace_the_error_the_run_failed_with(self,
storage):
+ run = DurableJournal(storage).start_run()
+ run.save(0, ENTRY)
+
+ with store_down():
+ run.fail(RuntimeError("the agent's own error"))
+
+
+class TestJournalThatIsNotAnObject:
+ @pytest.mark.parametrize(
+ "content",
+ [
+ pytest.param(b"{truncated", id="cut off mid-write"),
+ pytest.param(b'["not", "a", "journal"]', id="a JSON list"),
+ pytest.param(b"null", id="JSON null"),
+ ],
+ )
+ def test_is_handled_like_a_journal_cut_off_mid_write(self, storage,
content):
+ storage._get_path().write_bytes(content)
+
+ assert storage.load_step("k") is None
+ assert storage.save_step("k", ENTRY)
+ assert storage.load_step("k") == ENTRY
+
+
class TestMissingConfig:
def
test_a_missing_cache_path_names_the_option_without_naming_durable_true(self):
"""run_durably() callers reach this too, and they never pass
durable=True."""
