chore: fix polling event watcher self-unsubscribe deadlock (#33732)

## Summary & Motivation

Closes #5127.

Updated the polling watcher to stop invoking callbacks while holding
_callback_fn_list_lock, which removes the self-unsubscribe deadlock path
in
python_modules/dagster/dagster/_core/storage/event_log/polling_event_watcher.py.
The implementation now snapshots the callback list under lock and
dispatches outside it. I also re-enabled the skipped regression in
python_modules/dagster/dagster_tests/storage_tests/utils/event_log_storage.py.
## Test Plan

- python -m pytest
python_modules/dagster/dagster_tests/storage_tests/test_polling_event_watcher.py
- python -m pytest
python_modules/dagster/dagster_tests/storage_tests/test_event_log.py -k
'watch_unwatch and TestSqliteEventLogStorage'
- python -m pytest
python_modules/dagster/dagster_tests/storage_tests/test_event_log.py -k
'watch_unwatch and TestConsolidatedSqliteEventLogStorage'

## Changelog

> The changelog is generated by an agent that examines merged PRs and
> summarizes/categorizes user-facing changes. You can optionally replace
> this text with a terse description of any user-facing changes in your
PR,
> which the agent will prioritize. Otherwise, delete this section.

Author: cmpadden

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

@@ -157,6 +157,10 @@ def remove_callback(self, callback: Callable[[EventLogEntry, str], None]):
             if not self._callback_fn_list:
                 self._should_thread_exit.set()
 
+    def get_callbacks(self) -> list[CallbackAfterCursor]:
+        with self._callback_fn_list_lock:
+            return list(self._callback_fn_list)
+
     def run(self) -> None:
         """Polling function to update Observers with EventLogEntrys from Event Log DB.
         Wakes every POLLING_CADENCE &
@@ -177,15 +181,14 @@ def run(self) -> None:
             )
             cursor = conn.cursor
             for event_record in conn.records:
-                with self._callback_fn_list_lock:
-                    for callback_with_cursor in self._callback_fn_list:
-                        if (
-                            callback_with_cursor.cursor is None
-                            or EventLogCursor.parse(callback_with_cursor.cursor).storage_id()
-                            < event_record.storage_id
-                        ):
-                            callback_with_cursor.callback(
-                                event_record.event_log_entry,
-                                str(EventLogCursor.from_storage_id(event_record.storage_id)),
-                            )
+                for callback_with_cursor in self.get_callbacks():
+                    if (
+                        callback_with_cursor.cursor is None
+                        or EventLogCursor.parse(callback_with_cursor.cursor).storage_id()
+                        < event_record.storage_id
+                    ):
+                        callback_with_cursor.callback(
+                            event_record.event_log_entry,
+                            str(EventLogCursor.from_storage_id(event_record.storage_id)),
+                        )
             wait_time = INIT_POLL_PERIOD if conn.records else min(wait_time * 2, MAX_POLL_PERIOD)

python_modules/dagster/dagster_tests/storage_tests/utils/event_log_storage.py

@@ -2710,7 +2710,6 @@ def _throw(_x, _y):
         assert all([isinstance(event, dg.EventLogEntry) for event in event_list])
 
     # https://github.com/dagster-io/dagster/issues/5127
-    @pytest.mark.skip
     def test_watch_unwatch(self, storage):
         if not self.can_watch():
             pytest.skip("storage cannot watch runs")