@

perf(B04): 배수유역 세부 계산의 헛일 제거 — 282→154ms

DB 가 원격(dsm.chemifactory.com)이라 질의 하나가 곧 왕복 12ms 임을 실측.
순차 6건 130ms 중 대부분이 왕복 대기였음.

- context: 서로 기다릴 이유가 없는 질의를 두 묶음으로 asyncio.gather
  (묶음마다 자기 커넥션 — 풀 최대 20 이라 여유). 계획노선 읽기·지표면
  샘플러 열기도 같이 보냄.
- 격자 산출물(npz 3.1MB): 관을 옮길 때마다 다시 읽던 것을 파일 자국
  (수정시각·크기) 열쇠로 재사용. 분석이 다시 돌면 자국이 바뀌어 저절로 새로 읽음.

자체검증 — 용화 프로젝트(관 11·유역 11) 5회 중앙 282→154ms.
같은 입력에 응답 전체가 한 글자도 안 달라짐(JSON 정렬 비교).
전체 테스트 390 통과·17 건너뜀.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@

common_util/common_util_drainage_context.py

@@ -27,13 +27,13 @@ from B05_Profile.B05_Profile_Repository import (
     get_surface_crs_epsg,
 )
 from B06_Section.B06_Section_Repository import get_longitudinal_section
+from common_util.common_util_crs import resolve_project_crs
 from common_util.common_util_route_geometry import (
     RouteVertex,
     build_route_vertices,
     load_design_route,
 )
 from common_util.common_util_route_profile import Z_SOURCE_CSV, resolve_route_profile
-from common_util.common_util_crs import resolve_project_crs
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_surface_confirmation import get_surface_confirmation_params
 from common_util.common_util_surface_sampler import build_surface_sampler
@@ -63,37 +63,58 @@ class DrainageContext:
     surface_params: dict[str, Any] = field(default_factory=dict)
 
 
+async def _query(repository_call: Callable[..., Any], *args: Any) -> Any:
+    """저장소 함수 하나를 **자기 커넥션**으로 실행한다 — 같이 보내려면 커넥션이 갈려야 한다.
+
+    풀 최대치가 20이라 여기서 서너 개를 동시에 잡아도 여유가 있다(`config_system.DB_POOL_MAX`).
+    """
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        return await repository_call(connection, *args)
+
+
 async def load_drainage_context(project_id: UUID) -> tuple[DrainageContext | None, str]:
     """노선·종단 Z·좌표계를 준비한다. 실패하면 (None, 사용자에게 보일 사유).
 
     B05 확정 경로는 **있으면 쓰고 없으면 넘어간다** — B04는 WF1 화면이라 아직 경로가 없다.
     """
-    pool = get_db_pool()
-    async with pool.acquire() as connection:
-        stored_path = await get_project_storage_relative_path(connection, project_id)
-        if not stored_path:
-            return None, "프로젝트 저장 경로가 없습니다."
-        route = await get_latest_route(connection, project_id)
-        route_points: list[dict[str, Any]] = []
-        longitudinal: dict[str, Any] | None = None
-        if route:
-            route_points = await get_route_points(connection, int(route["id"]))
-            section = await get_longitudinal_section(connection, project_id, int(route["id"]))
-            longitudinal = (section or {}).get("data")
-        surface_model_id = (route or {}).get("surface_model_id")
-        db_epsg = await get_surface_crs_epsg(
-            connection, project_id, int(surface_model_id) if surface_model_id else 0
+    # DB 가 원격이라 질의 하나가 곧 왕복 12ms 다(2026-09-06 실측: 6건 순차 130ms).
+    # 서로 기다릴 이유가 없는 것끼리 묶어 두 묶음으로 보낸다 — 값은 그대로고 왕복만 겹친다.
+    stored_path, route, surface_params = await asyncio.gather(
+        _query(get_project_storage_relative_path, project_id),
+        _query(get_latest_route, project_id),
+        _query(get_surface_confirmation_params, str(project_id)),
+    )
+    if not stored_path:
+        return None, "프로젝트 저장 경로가 없습니다."
+
+    route_points: list[dict[str, Any]] = []
+    longitudinal: dict[str, Any] | None = None
+    surface_model_id = (route or {}).get("surface_model_id")
+    if route:
+        route_points, section, db_epsg = await asyncio.gather(
+            _query(get_route_points, int(route["id"])),
+            _query(get_longitudinal_section, project_id, int(route["id"])),
+            _query(
+                get_surface_crs_epsg,
+                project_id,
+                int(surface_model_id) if surface_model_id else 0,
+            ),
         )
-        surface_params = await get_surface_confirmation_params(connection, str(project_id))
+        longitudinal = (section or {}).get("data")
+    else:
+        db_epsg = await _query(get_surface_crs_epsg, project_id, 0)
 
     project_root = Path(resolve_stored_project_path(stored_path))
     # 설계 계통과 **같은 노선**을 쓴다 — 지표면 밖 구간을 자른 뒤의 노선이다. 원본을 그대로
     # 쓰면 유역·관이 확정 노선 밖에도 찍혀 종단 계획선이 그 관을 버린다(2026-09-01).
-    planned = await asyncio.to_thread(_read_planned_route, project_root, surface_params)
+    planned, sampler = await asyncio.gather(
+        asyncio.to_thread(_read_planned_route, project_root, surface_params),
+        asyncio.to_thread(_open_sampler, project_root, surface_params),
+    )
     if planned is None or len(planned.vertices) < 2:
         return None, "계획노선을 읽지 못했습니다. B03에서 노선 파일을 확인하세요."
 
-    sampler = await asyncio.to_thread(_open_sampler, project_root, surface_params)
     vertices, z_source = await asyncio.to_thread(
         resolve_route_profile,
         planned.vertices,

common_util/common_util_drainage_detail.py

@@ -69,6 +69,9 @@ from config.config_system import (
 
 logger = logging.getLogger(__name__)
 
+# 마지막으로 읽은 격자 산출물 — {npz 경로: (파일 자국, 읽은 결과)}. `read_road_routing` 참조.
+_routing_cache: dict[str, tuple[tuple[tuple[float, int], ...], "RoadRouting"]] = {}
+
 # 관 위치 선정 점수 배분 — 흐름 강도가 주, 종단 저점이 보조.
 _SCORE_WEIGHT_STRENGTH = 0.7
 _SCORE_WEIGHT_SAG = 0.3
@@ -230,12 +233,21 @@ def build_detail(
 
 
 def read_road_routing(directory: Path) -> RoadRouting | None:
-    """`03_road_routing` 산출물을 읽는다. 없으면 None."""
+    """`03_road_routing` 산출물을 읽는다. 없으면 None.
+
+    관을 하나 옮길 때마다 같은 파일(3.1MB)을 다시 읽어 45ms 를 썼다(2026-09-06 실측).
+    격자는 B04 분석이 다시 돌 때만 바뀌므로 **파일이 그대로면 앞서 읽은 것을 그대로 쓴다** —
+    파일 자국(수정시각·크기)이 열쇠라 분석이 다시 돌면 저절로 새로 읽는다.
+    """
     prefix = STAGES["road_routing"]
     array_path = directory / f"{prefix}_road_routing.npz"
     if not array_path.exists():
         logger.warning("배수유역: B04 분석 결과가 없습니다 (%s).", array_path)
         return None
+    cached = _routing_cache.get(str(array_path))
+    stamp = _file_stamp(array_path, directory / f"{prefix}_road_routing.geojson")
+    if cached is not None and cached[0] == stamp:
+        return cached[1]
     try:
         with np.load(array_path, allow_pickle=False) as data:
             spec = GridSpec(
@@ -266,9 +278,24 @@ def read_road_routing(directory: Path) -> RoadRouting | None:
         routing.road_cell_index.size,
         len(routing.base_pipes),
     )
+    # 프로젝트를 오가도 자국이 다르면 새로 읽으므로 한 벌만 들고 있으면 충분하다.
+    _routing_cache.clear()
+    _routing_cache[str(array_path)] = (stamp, routing)
     return routing
 
 
+def _file_stamp(*paths: Path) -> tuple[tuple[float, int], ...]:
+    """파일들의 (수정시각, 크기) — 하나라도 바뀌면 값이 달라진다. 없는 파일은 (0, 0)."""
+    marks = []
+    for path in paths:
+        try:
+            info = path.stat()
+            marks.append((info.st_mtime, info.st_size))
+        except OSError:
+            marks.append((0.0, 0))
+    return tuple(marks)
+
+
 def _read_geometry(path: Path, routing: RoadRouting) -> None:
     """계획도로선·2차 유역 외곽선·기본 관을 GeoJSON에서 읽어 채운다."""
     if not path.exists():