kaxil commented on code in PR #75916:
URL: https://github.com/apache/airflow/pull/75916#discussion_r4468545968
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -129,6 +180,9 @@ def load_step(self, key: str) -> dict[str, Any] | None:
def delete_steps(self, keys: Iterable[str]) -> None:
"""Remove these steps, and the journal file once nothing is left in
it."""
cache = self._load_cache()
+ if self._unreadable:
Review Comment:
If the first read on an attempt hits a transient error, `_unreadable` stays
set for the rest of the attempt. So when the attempt succeeds, none of the
cleanup calls (`DurableRun.cleanup`, `_forget_run`, `DurableJournal.cleanup`)
delete the failed attempt's journal. A later clear of the task then reads that
journal back and replays its steps and its `run_id`, while the `get_run_id`
docstring and the docs both say a clear after success starts fresh. Before this
change the attempt overwrote the file and cleanup removed it. Could
`delete_steps` retry the read here, or can `_unreadable` go?
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -91,10 +101,32 @@ def _load_cache(self) -> dict[str, Any]:
path = self._get_path()
try:
- self._cache = json.loads(path.read_text())
- except (FileNotFoundError, OSError, json.JSONDecodeError, ValueError):
+ text = path.read_text()
+ except FileNotFoundError:
self._cache = {}
+ return self._cache
+ except UnicodeDecodeError:
+ # Bytes that are not text cannot hold a journal; treat them like
any other invalid file.
+ text = ""
+ except Exception:
Review Comment:
On S3, a role without `s3:ListBucket` gets a 403 for a key that doesn't
exist yet, and s3fs raises that as `PermissionError`. Before this change that
was caught as an `OSError` and the journal started empty. Now it sets
`_unreadable`, so no step is ever written and the file is never created. Every
retry then hits the same 403, which leaves durability off for good, behind a
warning about a file "left in place" that isn't there. The task state store
backend doesn't guard reads at all, so I'd take your offer: keep the write
guard and the non-object JSON fix, and drop the read-side state. If the read
guard stays, can `PermissionError` count as missing?
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -107,20 +139,39 @@ def save_step(self, key: str, entry: dict[str, Any]) ->
bool:
"""
Store one journal step and rewrite the journal file.
- :return: ``True`` if the entry was written, ``False`` if it could not
be encoded.
+ :return: ``True`` if the entry was written, ``False`` if it was
skipped: it could
+ not be encoded, the write failed, or the journal file could not be
read.
"""
cache = self._load_cache()
+ if self._unreadable:
+ return False
+ previous = cache.get(key, _MISSING)
Review Comment:
What if the cache isn't touched until the write succeeds? `text =
json.dumps({**cache, key: entry})`, write it, then `cache[key] = entry`. That
drops `_MISSING` and `_restore`, along with its `self._cache is None` check,
which can't be hit since `_load_cache` always sets the cache first.
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -124,6 +124,11 @@ Without the option, the task fails with:
ValueError: durable=True requires [common.ai] durable_cache_path to be
set. Example: durable_cache_path = file:///tmp/airflow_durable_cache
+If the store fails while the task runs, the task does not. A journal file that
exists but
Review Comment:
This section covers both backends, but on 3.3+ a failed read from the task
state store still fails the task. Should this be scoped to the ObjectStorage
backend?
##########
providers/common/ai/tests/unit/common/ai/durable/test_storage.py:
##########
@@ -126,6 +127,96 @@ def
test_deleting_when_there_is_no_file_does_not_raise(self, storage):
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") as mock_log:
Review Comment:
Can the `log` and `read_text` patches here and below take `autospec=True`,
like `store_down()` does? `mock_log.warning.assert_called_once()` also passes
for either warning, so asserting the message would pin which branch fired.
--
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]
