Dev

app/agents/safety/berth_alerts.py

@@ -24,15 +24,8 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from .bulk_compatibility import build_bulk_conflicts
-from .graph_queries import (
-    find_adjacent_berth_conflicts,
-    find_bulk_exceptions,
-    find_bulk_group_conflicts,
-    find_bulk_groups,
-    find_imdg_no_segregation_required,
-    find_imdg_segregation_conflicts,
-    find_incompatible_conflicts,
-)
+from .graph_queries import find_adjacent_berth_conflicts
+from .graph_snapshot import get_snapshot
 from .rule_engine import (
     compute_bulk_compatibility_floor,
     compute_imdg_berth_adjacency_floor,
@@ -86,26 +79,22 @@
 async def _segregation_facts(
     driver: AsyncDriver, sem: asyncio.Semaphore, target_id: str, other_ids: list[str]
 ) -> tuple:
-    """대상 화물 하나의 혼재 판정에 필요한 그래프 조회 결과(판정은 하지 않는다)."""
-    async with sem:
-        return (
-            await find_incompatible_conflicts(
-                driver, target_chem_id=target_id, adjacent_chem_ids=other_ids
-            ),
-            await find_imdg_segregation_conflicts(
-                driver, target_chem_id=target_id, adjacent_chem_ids=other_ids
-            ),
-            await find_imdg_no_segregation_required(
-                driver, target_chem_id=target_id, adjacent_chem_ids=other_ids
-            ),
-            await find_bulk_group_conflicts(
-                driver, target_chem_id=target_id, adjacent_chem_ids=other_ids
-            ),
-            await find_bulk_groups(driver, chem_ids=[target_id, *other_ids]),
-            await find_bulk_exceptions(
-                driver, target_chem_id=target_id, adjacent_chem_ids=other_ids
-            ),
-        )
+    """대상 화물 하나의 혼재 판정에 필요한 그래프 조회 결과(판정은 하지 않는다).
+
+    [2026-09-30] Neo4j 조회 6개 → 메모리 사본(graph_snapshot). service.assess_safety 가 9/29 에
+    옮겨 간 것과 같은 사본·같은 답이다(tests/test_graph_snapshot_20260929.py). 운영(Aura)에서
+    대상 수십 개 × 조회 6개가 캐시 빈 첫 화면을 30초 붙잡았다. sem 은 테스트가 바꿔 끼우는
+    시그니처를 지키려고 남겨 둔다(쓰지 않음).
+    """
+    graph = await get_snapshot(driver)
+    return (
+        graph.incompatible_conflicts(target_id, other_ids),
+        graph.imdg_segregation_conflicts(target_id, other_ids),
+        graph.imdg_no_segregation_required(target_id, other_ids),
+        graph.bulk_group_conflicts(target_id, other_ids),
+        graph.bulk_groups([target_id, *other_ids]),
+        graph.bulk_exceptions(target_id, other_ids),
+    )
 
 
 async def _segregation_alerts(rows: list[dict], driver: AsyncDriver) -> list[dict]:
@@ -349,21 +338,15 @@ async def _adjacent_berth_alerts(rows: list[dict], driver: AsyncDriver) -> list[
     seen: set[frozenset] = set()
     alerts: list[dict] = []
 
-    # 조회는 동시에, 판정은 원래 순서대로(_segregation_alerts 와 같은 이유)
+    # 조회는 한 번에(대상 전부), 판정은 원래 순서대로(_segregation_alerts 와 같은 이유)
     targets = [
         (wharf, target)
         for wharf, cargos in by_wharf.items()
         for target in {c["chem_id"]: c for c in cargos}.values()
     ]
-    sem = asyncio.Semaphore(_NEO4J_CONCURRENCY)
-
-    async def _hits(wharf: str, target_id: str) -> list[dict]:
-        async with sem:
-            return await find_adjacent_berth_conflicts(
-                driver, wharf_name=wharf, target_chem_id=target_id, neighbor_cargo=cargo
-            )
-
-    all_hits = await asyncio.gather(*(_hits(w, t["chem_id"]) for w, t in targets))
+    all_hits = await find_adjacent_berth_conflicts(
+        driver, targets=[(w, t["chem_id"]) for w, t in targets], neighbor_cargo=cargo
+    )
 
     for (wharf, target), hits in zip(targets, all_hits):
         for hit in hits:

app/agents/safety/graph_queries.py

@@ -424,14 +424,20 @@ async def _tx(tx):
 #   가장 가까운 거리만 남긴다(min) — 같은 부두쌍이 여러 줄로 불어나지 않게.
 # ─────────────────────────────────────────────────────────────────────────────
 
+# [2026-09-30] 대상(부두·화물)마다 한 번씩 부르던 것을 대상 목록 한 번으로 묶었다 — 운영(Aura)은
+#   왕복이 약 300ms 라 대상 74개가 캐시 빈 관제 경고를 7.6초 붙잡았다. 대상 하나의 탐색·반환 행·
+#   정렬 키는 예전 그대로이고, 결과를 대상 순번(i)으로 나눠 돌려준다.
 _CYPHER_ADJACENT_BERTH_CONFLICTS = """
-MATCH (here:Berth {wharf_name: $wharf_name})-[adj:ADJACENT_TO]->(nb:Berth)
-WHERE nb.wharf_name <> $wharf_name
-WITH nb.wharf_name AS neighbor_wharf, min(adj.distance_m) AS distance_m
+UNWIND range(0, size($targets) - 1) AS i
+CALL (i) {
+WITH $targets[i].wharf_name AS wharf_name, $targets[i].target AS target
+MATCH (here:Berth {wharf_name: wharf_name})-[adj:ADJACENT_TO]->(nb:Berth)
+WHERE nb.wharf_name <> wharf_name
+WITH wharf_name, target, nb.wharf_name AS neighbor_wharf, min(adj.distance_m) AS distance_m
 UNWIND $neighbor_cargo AS nc
-WITH neighbor_wharf, distance_m, nc
+WITH wharf_name, target, neighbor_wharf, distance_m, nc
 WHERE nc.wharf_name = neighbor_wharf
-MATCH (a:Chemical {id: $target})
+MATCH (a:Chemical {id: target})
 MATCH (b:Chemical {id: nc.chem_id})
 // 충돌 근거 세 갈래를 한 서브쿼리로 합친다. 근거가 다른 신호라 어느 하나로
 // 대체할 수 없다 — MSDS 텍스트 마이닝 2방향(비대칭이라 양쪽을 다 봐야 한다.
@@ -462,7 +468,7 @@ async def _tx(tx):
     via                                  AS via,
     code                                 AS segregation_code,
     direction                            AS direction,
-    $wharf_name + ' —인접(' +
+    wharf_name + ' —인접(' +
         CASE WHEN distance_m IS NULL THEN '거리 미상'
              ELSE toString(toInteger(round(distance_m))) + 'm' END +
         ')→ ' + neighbor_wharf +
@@ -471,21 +477,25 @@ async def _tx(tx):
                    WHEN 'MSDS_INCOMPATIBLE' THEN '혼재금지(' + via + ')'
                    ELSE 'IMDG격리(' + via + ' → ' + coalesce(code, '?') + ')'
                END +
-        '→ ' + coalesce(a.name_ko, $target)  AS path_text
+        '→ ' + coalesce(a.name_ko, target)  AS path_text
+}
+RETURN i, neighbor_wharf, distance_m, chem_id, name_ko, callsgn, basis, via,
+       segregation_code, direction, path_text
 // 거리 미상(SK5~8 처럼 좌표 없이 부두번호로 이은 쌍)은 맨 뒤로. Cypher 는
 // ORDER BY ... NULLS LAST 를 안 받아서 정렬 키를 직접 만든다.
-ORDER BY coalesce(distance_m, 1000000.0), neighbor_wharf, chem_id
+ORDER BY i, coalesce(distance_m, 1000000.0), neighbor_wharf, chem_id
 """
 
 
 async def find_adjacent_berth_conflicts(
     driver: AsyncDriver,
     *,
-    wharf_name: str,
-    target_chem_id: str,
+    targets: list[tuple[str, str]],
     neighbor_cargo: list[dict],
-) -> list[dict]:
-    """대상 화물 하나에 대해 **인접 부두** 재항 화물과의 충돌을 한 번에 찾는다.
+) -> list[list[dict]]:
+    """대상 (부두, 화물) 각각에 대해 **인접 부두** 재항 화물과의 충돌을 한 번에 찾는다.
+
+    반환은 targets 와 같은 순서·같은 길이의 목록이고, 각 원소가 그 대상의 충돌 행이다.
 
     neighbor_cargo 는 호출부가 이미 읽어 둔 재항 현황 그대로 넘긴다 —
     각 항목은 최소한 {"wharf_name", "chem_id"} 를 갖고, "callsgn"·"cargo_name"
@@ -494,8 +504,15 @@ async def find_adjacent_berth_conflicts(
     반환 행에는 `path_text`(그래프가 만든 경로 문장)가 들어 있다. 화면·프롬프트는
     이 문장을 그대로 쓰면 되고, 근거를 다시 조립하지 않는다.
     """
-    if not wharf_name or not target_chem_id or not neighbor_cargo:
-        return []
+    out: list[list[dict]] = [[] for _ in targets]
+    # 부두나 화물이 빈 대상은 예전처럼 조회 없이 빈 결과
+    query_targets = [
+        {"wharf_name": w, "target": t, "idx": n}
+        for n, (w, t) in enumerate(targets)
+        if w and t
+    ]
+    if not query_targets or not neighbor_cargo:
+        return out
 
     payload = [
         {
@@ -508,17 +525,19 @@ async def find_adjacent_berth_conflicts(
         if c.get("wharf_name") and c.get("chem_id")
     ]
     if not payload:
-        return []
+        return out
 
     async with driver.session() as session:
 
         async def _tx(tx):
             result = await tx.run(
                 _CYPHER_ADJACENT_BERTH_CONFLICTS,
-                wharf_name=wharf_name,
-                target=target_chem_id,
+                targets=query_targets,
                 neighbor_cargo=payload,
             )
             return [record.data() async for record in result]
 
-        return await session.execute_read(_tx)
+        rows = await session.execute_read(_tx)
+    for row in rows:
+        out[query_targets[row.pop("i")]["idx"]].append(row)
+    return out

app/api/v1/arrivals.py

@@ -32,12 +32,29 @@
                pv.agency_name, pv.prev_port_nm, pv.entry_purpose_nm,
                pv.arrival_at_utc, pv.departure_sched_utc, pv.arrival_report_type,
                pv.arrival_facility_cd, pv.arrival_facility_nm, pv.collected_at_utc,
-               pv.entry_year, pv.entry_count
+               pv.entry_year, pv.entry_count,
+               CASE WHEN pv.entry_year IS NOT NULL AND pv.entry_count IS NOT NULL
+                    THEN upper(btrim(pv.callsgn)) || '_' || pv.entry_year::text || '_' || lpad(pv.entry_count::text, 3, '0')
+               END AS port_call_key
         FROM portmis_vessel pv
         WHERE pv.is_liquid_cargo_vessel
           AND pv.arrival_at_utc BETWEEN now() - make_interval(hours => :past_hours)
                                     AND now() + make_interval(hours => :ahead_hours)
     ),
+    -- [2026-09-25] 이 입항 건(콜사인·입항연도·입항횟수)에 단 화물 전부. 입항 후
+    -- 위치 화면(mart.vessel_current_call 경유)과 같은 키라 같은 화물이 나온다.
+    -- [2026-09-30] 행마다 LATERAL 로 cargo_msds 를 다시 훑던 것(100건 × 23ms)을 한 번 집계로 바꿈.
+    cg AS (
+        SELECT cm.port_call_key,
+               json_agg(json_build_object(
+                   'name', coalesce(cm.msds_name_ko, cm.cargo_name_raw),
+                   'un_no', cm.dg_un_no, 'chem_id', cm.chem_id,
+                   'cas_no', cm.cas_no, 'is_synthetic', cm.is_synthetic
+               ) ORDER BY (cm.chem_id IS NULL), cm.bl_no) AS cargos
+        FROM mart.cargo_msds cm
+        WHERE cm.port_call_key IN (SELECT port_call_key FROM arr)
+        GROUP BY cm.port_call_key
+    ),
     fac AS (
         SELECT DISTINCT ON (source_name) source_name, wharf_name, facility_type
         FROM mart.facility_alias
@@ -85,30 +102,18 @@
            a.collected_at_utc,
            cg.cargos,
            -- [2026-09-29] 이 행의 입항 건 키와 판정 잡이 지금 보는 입항 건 키. 판정 기록은 같은 입항 건 것만 붙인다.
-           CASE WHEN a.entry_year IS NOT NULL AND a.entry_count IS NOT NULL
-                THEN upper(btrim(a.callsgn)) || '_' || a.entry_year::text || '_' || lpad(a.entry_count::text, 3, '0')
-           END AS port_call_key,
-           (SELECT vc.port_call_key FROM mart.vessel_current_call vc
-            WHERE vc.callsgn = upper(btrim(a.callsgn))) AS current_call_key
+           a.port_call_key,
+           vc.port_call_key AS current_call_key
     FROM arr a
     LEFT JOIN fac f ON f.source_name = a.arrival_facility_nm
     LEFT JOIN berth b ON b.wharf_name = f.wharf_name
     LEFT JOIN pos p ON p.cs = upper(btrim(a.callsgn))
     LEFT JOIN draught d ON d.cs = upper(btrim(a.callsgn))
     LEFT JOIN spec s ON s.cs = upper(btrim(a.callsgn))
-    -- [2026-09-25] 이 입항 건(콜사인·입항연도·입항횟수)에 단 화물 전부. 입항 후
-    -- 위치 화면(mart.vessel_current_call 경유)과 같은 키라 같은 화물이 나온다.
-    LEFT JOIN LATERAL (
-        SELECT json_agg(json_build_object(
-                   'name', coalesce(cm.msds_name_ko, cm.cargo_name_raw),
-                   'un_no', cm.dg_un_no, 'chem_id', cm.chem_id,
-                   'cas_no', cm.cas_no, 'is_synthetic', cm.is_synthetic
-               ) ORDER BY (cm.chem_id IS NULL), cm.bl_no) AS cargos
-        FROM mart.cargo_msds cm
-        WHERE a.entry_year IS NOT NULL AND a.entry_count IS NOT NULL
-          AND cm.port_call_key = upper(btrim(a.callsgn)) || '_' || a.entry_year::text
-                                 || '_' || lpad(a.entry_count::text, 3, '0')
-    ) cg ON true
+    LEFT JOIN cg ON cg.port_call_key = a.port_call_key
+    -- [2026-09-30] 상관 서브쿼리로 행마다 뷰를 다시 계산하던 것(100건 × 20ms)을 조인으로 바꿈.
+    --   vessel_current_call 은 콜사인당 1행(스칼라 서브쿼리였으므로)이라 행 수가 늘지 않는다.
+    LEFT JOIN mart.vessel_current_call vc ON vc.callsgn = upper(btrim(a.callsgn))
     ORDER BY a.arrival_at_utc
 """)
 

app/database.py

@@ -6,7 +6,14 @@
 
 settings = get_settings()
 
-engine = create_async_engine(settings.database_url, echo=False, future=True)
+# JIT 끔: mart 뷰 조인의 행 수 과대추정(실제 1.7천 → 추정 477만)으로 비용이 JIT 기준을
+# 넘어, 0.2초 쿼리에 LLVM 컴파일 3초가 붙었다(2026-09-30 로컬 PG16 실측). RDS 는 기본이 off.
+engine = create_async_engine(
+    settings.database_url,
+    echo=False,
+    future=True,
+    connect_args={"server_settings": {"jit": "off"}},
+)
 AsyncSessionFactory = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)