Order RUN_FAILURE_REASON_TAG write before run status update (#24730)

## Summary & Motivation

In `SqlRunStorage.handle_run_event`, the PIPELINE_FAILURE handler used
to write the run status update and the `dagster/failure_reason` tag in
**two separate transactions**:

1. `with self.connect() as conn:` — UPDATE `runs.status = FAILURE`
(commits on exit).
2. `self.add_run_tags(run_id, {RUN_FAILURE_REASON_TAG: ...})` — opens
its own connection.

A caller polling for `run.is_finished` could observe `status=FAILURE`
between those two writes and then read `run.tags` without the tag —
`KeyError: 'dagster/failure_reason'`.

This is the source of the flake in
`test_retry_failure_all_steps_with_reexecution_params` ([example
failure](https://buildkite.com/dagster/internal/builds/152044#019e4b21-52ee-4964-807d-47d28eed96a5)):
the test calls `wait_for_runs_to_finish` (status-only) then reads the
tag immediately.

## Fix

- Compute the `failure_reason` tag value upfront and include it in the
`run_body` serialized into the runs UPDATE.
- Inside the same `with self.connect() as conn:` block, write the
`RunTagsTable` row **first**, then the runs UPDATE.
- On `SqliteRunStorage`, `connect()` wraps in `conn.begin()` so the two
writes commit as a unit.
- On `PostgresRunStorage`, the engine is configured with
`isolation_level="AUTOCOMMIT"`, so each `conn.execute` commits
independently. Ordering the tag write first ensures that by the time a
reader observes the new status, the tag row is already there.
- `update_timestamp` is written exactly once on the runs row (using the
event timestamp), avoiding the regression that calling `add_run_tags`
separately would introduce (where `runs.update_timestamp` could be
written as `now()` and then overwritten with an earlier event timestamp,
breaking `RunsFilter(updated_after=...)` cursors).
- Use the in-memory `run.tags` to choose UPDATE vs INSERT against
`RunTagsTable`, mirroring `add_run_tags` — no extra SELECT round-trip.

## Test Plan

- [x] `tox -e py312-storage_tests -- -k test_failure_event_updates_tags`
— all 3 variants (Sqlite, InMemory, Legacy) pass.
- [x] Full `dagster_tests/storage_tests/test_run_storage.py` suite —
1007 passed, 105 pre-existing skips.
- [ ] CI green on `dagster-graphql` suites that previously surfaced the
flake.

## Changelog

Eliminate a race where readers polling for a run to reach FAILURE could
miss the `dagster/failure_reason` tag.

Internal-RevId: 2cb1d47fe55e3b99a82dbbd2ed244046f1dfbc36

Author: smackesey

python_modules/dagster/dagster/_core/storage/runs/sql_run_storage.py

@@ -188,6 +188,20 @@ def handle_run_event(
 
         new_job_status = EVENT_TYPE_TO_PIPELINE_RUN_STATUS[event.event_type]
 
+        # Threaded into the run_body and the run_tags write below so readers that
+        # observe a finished status are guaranteed to also see this tag.
+        updated_run = run.with_status(new_job_status)
+        failure_reason_tag_value: str | None = None
+        if event.event_type == DagsterEventType.PIPELINE_FAILURE and isinstance(
+            event.event_specific_data, JobFailureData
+        ):
+            failure_reason = event.event_specific_data.failure_reason
+            if failure_reason and failure_reason != RunFailureReason.UNKNOWN:
+                failure_reason_tag_value = failure_reason.value
+                updated_run = updated_run.with_tags(
+                    merge_dicts(run.tags, {RUN_FAILURE_REASON_TAG: failure_reason_tag_value})
+                )
+
         run_stats_cols_in_index = self.has_run_stats_index_cols()
 
         kwargs = {}
@@ -208,25 +222,42 @@ def handle_run_event(
             kwargs["end_time"] = update_timestamp.timestamp()
 
         with self.connect() as conn:
+            # Write the run_tags row before the runs update. PostgresRunStorage uses
+            # an AUTOCOMMIT engine, so each `conn.execute` commits independently;
+            # writing the tag first ensures that any reader that observes the new
+            # status sees the tag too.
+            if failure_reason_tag_value is not None:
+                if RUN_FAILURE_REASON_TAG in run.tags:
+                    conn.execute(
+                        RunTagsTable.update()
+                        .where(
+                            db.and_(
+                                RunTagsTable.c.run_id == run_id,
+                                RunTagsTable.c.key == RUN_FAILURE_REASON_TAG,
+                            )
+                        )
+                        .values(value=failure_reason_tag_value)
+                    )
+                else:
+                    conn.execute(
+                        RunTagsTable.insert().values(
+                            run_id=run_id,
+                            key=RUN_FAILURE_REASON_TAG,
+                            value=failure_reason_tag_value,
+                        )
+                    )
+
             conn.execute(
                 RunsTable.update()
                 .where(RunsTable.c.run_id == run_id)
                 .values(
-                    run_body=serialize_value(run.with_status(new_job_status)),
+                    run_body=serialize_value(updated_run),
                     status=new_job_status.value,
                     update_timestamp=update_timestamp,
                     **kwargs,
                 )
             )
 
-        if event.event_type == DagsterEventType.PIPELINE_FAILURE and isinstance(
-            event.event_specific_data, JobFailureData
-        ):
-            failure_reason = event.event_specific_data.failure_reason
-
-            if failure_reason and failure_reason != RunFailureReason.UNKNOWN:
-                self.add_run_tags(run_id, {RUN_FAILURE_REASON_TAG: failure_reason.value})
-
     def _row_to_run(self, row: dict) -> DagsterRun:
         run = deserialize_value(row["run_body"], DagsterRun)
         status = DagsterRunStatus(row["status"])