kaxil commented on code in PR #76381:
URL: https://github.com/apache/airflow/pull/76381#discussion_r4467681495
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -92,7 +92,14 @@ 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):
+ except FileNotFoundError:
+ self._cache = {}
+ except OSError:
+ log.warning(
+ "Durable: could not load journal; stopping task to avoid
replaying steps", exc_info=True
+ )
+ raise
+ except (json.JSONDecodeError, ValueError):
Review Comment:
A truncated journal still lands here as empty, and the next `save_step`
overwrites it. `write_text` isn't atomic on `file://` or NFS, so a worker
killed mid-write leads to the same duplicate calls this PR is about, with
nothing in the log. Should this at least log a warning? `read_state` in
`batch/state.py` raises on a corrupt file for this reason.
##########
providers/common/ai/tests/unit/common/ai/durable/test_storage.py:
##########
@@ -82,6 +82,13 @@ def test_a_retry_reads_what_the_previous_attempt_wrote(self,
storage):
def test_missing_key_is_none(self, storage):
assert storage.load_step("missing") is None
+ @patch.object(ObjectStoragePath, "read_text", autospec=True,
side_effect=OSError("S3 is unavailable"))
+ def
test_a_storage_read_failure_stops_the_task_before_it_can_replay_live(self, _,
storage, caplog):
+ with pytest.raises(OSError, match="S3 is unavailable"):
+ storage.load_step("k")
+
+ assert "Durable: could not load journal; stopping task to avoid
replaying steps" in caplog
Review Comment:
This `in caplog` raises `TypeError` on the 2.11.1 and 3.0.6 compat jobs.
Without `airflow.sdk._shared` the `caplog` override hands back pytest's stock
`LogCaptureFixture`, which doesn't support `in`, and those are exactly the
versions this backend is for. Can we drop the log assertion and the `caplog`
arg? The `pytest.raises` already pins the behaviour.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -92,7 +92,14 @@ 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):
+ except FileNotFoundError:
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 the
first attempt treated it as an empty journal and created the file; now every
attempt fails before anything is written. Failing closed still seems right
since a denied read of an existing journal looks the same, but can
`durable_execution.rst` say the role needs `s3:ListBucket`?
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -92,7 +92,14 @@ 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):
+ except FileNotFoundError:
+ self._cache = {}
+ except OSError:
+ log.warning(
+ "Durable: could not load journal; stopping task to avoid
replaying steps", exc_info=True
+ )
+ raise
Review Comment:
This now also runs on the read right after `delete_steps` unlinks the file
and resets `_cache` to `None`. With two durable agent runs in one `@task`, run
1's cleanup deletes the journal and run 2's `start_run` reads storage again
through `save_step`. A transient error on that read fails the task, and the
retry runs agent 1 live again, which is what this change is meant to prevent.
Setting `self._cache = {}` after the unlink (and in its `FileNotFoundError`
branch) skips that read and keeps `delete_steps` from raising through
`_load_cache`.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/storage.py:
##########
@@ -92,7 +92,14 @@ 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):
+ except FileNotFoundError:
+ self._cache = {}
+ except OSError:
+ log.warning(
+ "Durable: could not load journal; stopping task to avoid
replaying steps", exc_info=True
Review Comment:
In this package "replay" means serving a recorded step, which is what we
want a retry to do, so this message reads backwards. The risk is steps an
earlier attempt finished running again live. `exc_info=True` plus the bare
`raise` also prints the traceback twice in the task log. Something like
`"Durable: could not read the journal; failing this attempt so finished steps
don't run again"`, without `exc_info`?
--
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]
