Ensure only one thread is ever building the full asset graph at once (#21570)

Summary:
A better version of https://github.com/dagster-io/internal/pull/21249 - instead of doing it in one specific context, do it anytime any context is constructing the full asset graph (since it is expensive and we would prefer a bit of contention to having multiple thread doing a heavy asset graph simultaneously)

Test Plan:
BK

## Summary & Motivation

## Test Plan

## 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.

Internal-RevId: 19ac17aaed30d702e4c104dc55882ea7dc45b3f5

Author: gibsondan

python_modules/dagster/dagster/_core/workspace/workspace.py

@@ -1,6 +1,6 @@
+import threading
 from collections.abc import Mapping
 from enum import Enum
-from functools import cached_property
 from typing import TYPE_CHECKING, Annotated
 
 from dagster._record import ImportFrom, record
@@ -54,17 +54,28 @@ class CodeLocationStatusEntry:
     has_load_error: bool
 
 
-@record
 class CurrentWorkspace:
-    code_location_entries: Mapping[str, CodeLocationEntry]
+    def __init__(self, code_location_entries: Mapping[str, CodeLocationEntry]):
+        self.code_location_entries = code_location_entries
+        self._asset_graph_lock = threading.Lock()
+        self._asset_graph: RemoteWorkspaceAssetGraph | None = None
 
-    @cached_property
+    @property
     def asset_graph(self) -> "RemoteWorkspaceAssetGraph":
+        if self._asset_graph is not None:
+            return self._asset_graph
+
         from dagster._core.definitions.assets.graph.remote_asset_graph import (
             RemoteWorkspaceAssetGraph,
         )
 
-        return RemoteWorkspaceAssetGraph.build(self)
+        # building the full asset graph is expensive - so ensure at most one
+        # thread is doing that work at once, and subsequent threads reuse the result
+        with self._asset_graph_lock:
+            if self._asset_graph is not None:
+                return self._asset_graph
+            self._asset_graph = RemoteWorkspaceAssetGraph.build(self)
+            return self._asset_graph
 
     def with_code_location(self, name: str, entry: CodeLocationEntry) -> "CurrentWorkspace":
         return CurrentWorkspace(code_location_entries={**self.code_location_entries, name: entry})