Fix SqlPollingEventWatcher.close() deadlock when callback re-enters unwatch_run (#25432)

## Summary & Motivation

Fixes a teardown deadlock in `SqlPollingEventWatcher.close()` that
surfaced as `Failed: Timeout (>240.0s) from pytest-timeout` in
`TestPostgresEventLogStorage.test_watch_unwatch` ([master build
155280](https://buildkite.com/dagster/internal/builds/155280#019eacad-a73d-4453-b8f0-1ce503877bac)).

PR #33732 (`8750f882227`, May 27) fixed the original
`_callback_fn_list_lock` self-deadlock by snapshotting the callback list
before firing callbacks outside the lock, and re-enabled the
previously-skipped `test_watch_unwatch` regression. That fix exposed a
second, distinct deadlock cycle on `_dict_lock`:

1. A watcher thread fires a user callback (now outside any lock).
2. The callback — e.g. the test's `_unsub` at
`event_log_storage.py:2724` — calls `storage.end_watch(...)` →
`SqlPollingEventWatcher.unwatch_run()` → blocks on `with
self._dict_lock`.
3. Concurrently, the test's `instance_for_test` exits and runs
`dispose()` → `close()`, which grabs `_dict_lock` first.
4. Inside the lock, `close()` calls `join()` on each watcher thread. The
watcher is blocked on `_dict_lock`. The main thread waits forever on
`join()`.

The Buildkite timeout dump shows the hung thread parked at
`polling_event_watcher.py:82` inside `unwatch_run`, called from
`_unsub`, which matches this cycle exactly. The test body passed
(279/279); only teardown deadlocked.

## Fix

Hold `_dict_lock` only long enough to snapshot the watcher list, set
`should_thread_exit`, and clear the dict. Then `join()` the snapshot
**outside** the lock. The thread's pending `unwatch_run()` call can now
acquire the lock briefly (finds no entry, since the dict was cleared)
and return, letting the thread exit cleanly and `join()` complete.

```python
def close(self) -> None:
    if not self._disposed:
        self._disposed = True
        with self._dict_lock:
            watcher_threads = list(self._run_id_to_watcher_dict.values())
            for watcher_thread in watcher_threads:
                if not watcher_thread.should_thread_exit.is_set():
                    watcher_thread.should_thread_exit.set()
            self._run_id_to_watcher_dict = {}
        # Join outside _dict_lock — see comment.
        for watcher_thread in watcher_threads:
            watcher_thread.join()
```

`self._disposed = True` is set before the lock release, so any racing
`watch_run` call still trips the `not self._disposed` invariant check.

## Test plan

- [x] `just check_ruff` clean
- [x] `just ty` clean (4775 files, 0 errors)
- [ ] Buildkite green on the existing `test_watch_unwatch` regression
across the SQL polling-watcher matrix (sqlite, consolidated-sqlite,
postgres). This is the regression test re-enabled by #33732 — it was
passing flakily before this fix; it should now be stable.

Internal-RevId: 37a56878c5eb49052d1571ce7dbfe758043b823d

Author: smackesey

python_modules/dagster/dagster/_core/storage/event_log/polling_event_watcher.py

@@ -89,12 +89,16 @@ def close(self) -> None:
         if not self._disposed:
             self._disposed = True
             with self._dict_lock:
-                for watcher_thread in self._run_id_to_watcher_dict.values():
+                watcher_threads = list(self._run_id_to_watcher_dict.values())
+                for watcher_thread in watcher_threads:
                     if not watcher_thread.should_thread_exit.is_set():
                         watcher_thread.should_thread_exit.set()
-                for run_id in self._run_id_to_watcher_dict:
-                    self._run_id_to_watcher_dict[run_id].join()
                 self._run_id_to_watcher_dict = {}
+            # Join outside _dict_lock — a watcher thread may be running a callback that
+            # re-enters unwatch_run() and tries to acquire _dict_lock; joining under the
+            # lock would deadlock.
+            for watcher_thread in watcher_threads:
+                watcher_thread.join()
 
 
 class SqlPollingRunIdEventWatcherThread(threading.Thread):