perf(342-2): [확정] · [저장] 서버 재계산 입력 지문 · 바뀐 행만 씀 · 행 읽기 한 번

- 지문 = Node 입력 전부 + 번들 판 · 같은 입력이면 Node 건너뜀(틀 = B07 덧그림 기억 · TS 가 번들보다 새로우면 버림)
- 측점 행 쓰기: 여러 벌을 읽기 한 번 · 바뀐 행만 한 문장(merge_cross_section_design_batches) · 앞 벌의 새 행을 뒤 벌이 찾음
- 종단 다시 읽기 지움(병렬 조회 한 번) · 유토곡선 같은 값이면 안 씀 · Node 결과 사본에 balloon 위치
- 시험 test_342_2_confirm_fingerprint

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD

B06_Section/B06_Section_Repository_Bulk.py

@@ -42,7 +42,23 @@ async def merge_cross_section_designs(
     행이 없는 측점은 `project_id` 가 오면 새로 만든다 — 구조물(비정규) 측점은 B05 확정이
     파일만 쓰고 DB 행을 안 만들기 때문이다(한 행짜리 함수와 같은 규칙).
     """
-    if not entries:
+    return await merge_cross_section_design_batches(
+        connection, route_id=route_id, batches=[(entries, replace, project_id)]
+    )
+
+
+async def merge_cross_section_design_batches(
+    connection: aiomysql.Connection,
+    *,
+    route_id: int,
+    batches: list[tuple[list[tuple[float, dict[str, Any]]], bool, UUID | None]],
+) -> int:
+    """여러 벌(`(entries, replace, project_id)`)을 차례로 얹되 읽기 한 번 · 쓰기 한 번(342-2).
+
+    앞 벌이 만든 새 행도 뒤 벌이 찾는다(따로 부르던 때와 같은 결과). 값이 그대로인 행은 쓰지
+    않는다 — 같은 입력의 [저장]·[확정]이 행을 다시 쓰지 않게.
+    """
+    if not any(entries for entries, _, _ in batches):
         return 0
 
     async with connection.cursor() as cursor:
@@ -56,8 +72,11 @@ async def merge_cross_section_designs(
     for row_id, _chainage, raw in rows:
         data = json.loads(raw) if isinstance(raw, str) else raw
         existing[int(row_id)] = data if isinstance(data, dict) else {}
+    original = dict(existing)  # 행 dict 는 제자리에서 안 고침(아래가 늘 사본을 만듦) — 대조 기준
     # 같은 측점이 여러 행이면 뒤에 온 것(=큰 id)을 쓴다 — 한 행짜리 함수의 `ORDER BY id DESC`.
     ordered = [(float(chainage), int(row_id)) for row_id, chainage, _ in rows]
+    # 새 행 = 음수 id · 뒤 벌이 찾음(있는 행이 먼저 · 새 행끼리는 먼저 만든 것) · 끝에 INSERT
+    fresh: dict[int, tuple[str, float]] = {}
 
     def find(chainage_m: float) -> int | None:
         best: int | None = None
@@ -68,40 +87,44 @@ async def merge_cross_section_designs(
 
     # 행마다 한 벌 — 같은 행에 여러 줄이 오면 앞 줄 결과 위에 얹음(`CASE id` 는 첫 WHEN 만 씀 ·
     # 318-4 옛 조정값 비움 뒤 새 값)
-    updates: dict[int, str] = {}
-    inserts: list[tuple[str, int, float, str]] = []
-    for chainage_m, payload in entries:
-        if not payload and not replace:
-            continue
-        # 얹기의 None = 그 열쇠 지움(JSON merge-patch · 318-4 관 종류 바꿈 옛 조정값)
-        drop = [] if replace else [key for key, value in payload.items() if value is None]
-        if drop:
-            payload = {key: value for key, value in payload.items() if value is not None}
-        row_id = find(chainage_m)
-        if row_id is None:
-            if project_id is None or not (payload or replace):
+    for entries, replace, project_id in batches:
+        for chainage_m, payload in entries:
+            if not payload and not replace:
                 continue
-            inserts.append(
-                (
-                    str(project_id),
-                    route_id,
-                    chainage_m,
-                    json.dumps({"design": payload}, ensure_ascii=False),
-                )
-            )
-            continue
-        data = dict(existing[row_id])
-        if replace:
-            data["design"] = payload
-        else:
-            design = data.get("design")
-            design = dict(design) if isinstance(design, dict) else {}
-            design.update(payload)
-            for key in drop:
-                design.pop(key, None)
-            data["design"] = design
-        existing[row_id] = data
-        updates[row_id] = json.dumps(data, ensure_ascii=False)
+            # 얹기의 None = 그 열쇠 지움(JSON merge-patch · 318-4 관 종류 바꿈 옛 조정값)
+            drop = [] if replace else [key for key, value in payload.items() if value is None]
+            if drop:
+                payload = {key: value for key, value in payload.items() if value is not None}
+            row_id = find(chainage_m)
+            if row_id is None:
+                if project_id is None or not (payload or replace):
+                    continue
+                row_id = -(len(fresh) + 1)
+                fresh[row_id] = (str(project_id), chainage_m)
+                ordered.append((chainage_m, row_id))
+                existing[row_id] = {"design": payload}
+                continue
+            data = dict(existing[row_id])
+            if replace:
+                data["design"] = payload
+            else:
+                design = data.get("design")
+                design = dict(design) if isinstance(design, dict) else {}
+                design.update(payload)
+                for key in drop:
+                    design.pop(key, None)
+                data["design"] = design
+            existing[row_id] = data
+
+    updates = {
+        row_id: json.dumps(data, ensure_ascii=False)
+        for row_id, data in existing.items()
+        if row_id > 0 and data != original[row_id]
+    }
+    inserts = [
+        (owner, route_id, chainage_m, json.dumps(existing[row_id], ensure_ascii=False))
+        for row_id, (owner, chainage_m) in fresh.items()
+    ]
 
     written = 0
     async with connection.cursor() as cursor:

B06_Section/B06_Section_Server_Calc_Prebuild.py

@@ -23,6 +23,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 import logging
 import time
@@ -36,8 +37,8 @@ from B06_Section.B06_Section_Repository import (
     get_longitudinal_section,
     merge_longitudinal_section_data,
 )
-from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
-from common_util.common_util_node_bundle import run_bundle_json
+from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_design_batches
+from common_util.common_util_node_bundle import bundle_stale, run_bundle_json
 from common_util.common_util_project_settings import (
     haul_equipment_limits,
     mixed_conversion_factors,
@@ -73,6 +74,37 @@ _AREA_KEYS = (
 )
 
 
+# 입력 지문 → Node 결과(342-2 · 틀 = B07 덧그림 기억) — 같은 입력이면 Node 를 다시 안 돌림.
+# 지문 = Node 에 넘기는 입력 **전부** + 번들 판(mtime) · TS 가 번들보다 새로우면 버림.
+# ponytail: 프로세스 메모리 · 최근 몇 벌만. 서버 여러 대면 `_derived/` 파일로.
+_NODE_CACHE: dict[str, Any] = {}
+_NODE_CACHE_LIMIT = 8
+
+
+def _node_fingerprint(node_input: dict[str, Any]) -> str:
+    """Node 입력 전부 + 번들 판 → 지문. 입력 한 칸이라도 바뀌면 다른 값."""
+    digest = hashlib.sha1(
+        json.dumps(node_input, sort_keys=True, default=str).encode("utf-8"),
+        usedforsecurity=False,
+    )
+    digest.update(str(BUNDLE.stat().st_mtime_ns if BUNDLE.is_file() else 0).encode())
+    return digest.hexdigest()
+
+
+def _node_output(node_input: dict[str, Any]) -> Any:
+    """같은 입력이면 지난 결과 · 아니면 Node(면적 · 유토곡선)를 돌려 기억한다."""
+    key = _node_fingerprint(node_input)
+    if key in _NODE_CACHE and not bundle_stale(BUNDLE):
+        return _NODE_CACHE[key]
+    output = run_bundle_json(BUNDLE, _NPM_SCRIPT, node_input)
+    if output is None:  # 실패는 기억하지 않음 — 다음 [저장]이 다시 셈
+        return None
+    if len(_NODE_CACHE) >= _NODE_CACHE_LIMIT:
+        _NODE_CACHE.pop(next(iter(_NODE_CACHE)))
+    _NODE_CACHE[_node_fingerprint(node_input)] = output  # 셈 사이 번들을 다시 지었으면 새 판으로
+    return output
+
+
 async def haul_inputs_for(project_id: Any) -> dict[str, Any]:
     """B08 이 낸 **구조물 몫**(채집석 공제·구조물 잔토)을 받아 온다.
 
@@ -268,9 +300,7 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int:
     marks.append(("포장·세월교 보정", time.perf_counter()))
 
     output = await asyncio.to_thread(
-        run_bundle_json,
-        BUNDLE,
-        _NPM_SCRIPT,
+        _node_output,
         {
             "detail": detail,
             "context": _mass_haul_context(
@@ -303,43 +333,42 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int:
     if not fixed and not rows and not mass_haul and not extra_walls:
         return 0
 
+    stored = (longitudinal_row or {}).get("data") or {}
+    if isinstance(mass_haul, dict):
+        # balloon 위치는 **사용자가 끌어 옮긴 화면값**이다 — 서버가 만들지 않으므로
+        # 저장분에서 떼어 새 유토곡선에 도로 붙인다(2026-09-06). 위 병렬 조회 한 번을 씀(342-2).
+        offsets = (stored.get("mass_haul") or {}).get("balloon_offsets")
+        if offsets is not None:  # 사본에 붙임 — Node 결과는 지문 기억이 쥐고 있음
+            mass_haul = {**mass_haul, "balloon_offsets": offsets}
+        if mass_haul == stored.get("mass_haul"):
+            mass_haul = None  # 같은 값 — 다시 안 씀
+
+    area_entries: list[tuple[float, dict[str, Any]]] = []
+    for row in rows if isinstance(rows, list) else []:
+        patch: dict[str, Any] = {
+            key: float(row[key]) for key in _AREA_KEYS if isinstance(row.get(key), (int, float))
+        }
+        if patch:
+            area_entries.append((float(row["chainage_m"]), patch))
+
     updated = 0
     async with pool.acquire() as connection:
-        # balloon 위치는 **사용자가 끌어 옮긴 화면값**이다 — 서버가 만들지 않으므로
-        # 저장분에서 떼어 새 유토곡선에 도로 붙인다(2026-09-06).
-        if isinstance(mass_haul, dict):
-            existing = await get_longitudinal_section(connection, project_uuid, route_id)
-            stored = (existing or {}).get("data") or {}
-            offsets = (stored.get("mass_haul") or {}).get("balloon_offsets")
-            if offsets is not None:
-                mass_haul["balloon_offsets"] = offsets
-
         await connection.begin()
         try:
-            # 행마다 쓰면 원격 DB 왕복이 행 수만큼 난다(측정: 22행 670ms) — 한 문장으로 묶는다.
-            await merge_cross_section_designs(
+            # 행 읽기 한 번 · 바뀐 행만 한 문장(342-2) — 보정 설계 갈아 끼움 → 면적 · 벽 얹기 차례.
+            updated = await merge_cross_section_design_batches(
                 connection,
                 route_id=route_id,
-                entries=[(float(item.get("chainage_m") or 0.0), item["design"]) for item in fixed],
-                replace=True,
-                project_id=project_uuid,
+                batches=[
+                    (
+                        [(float(item.get("chainage_m") or 0.0), item["design"]) for item in fixed],
+                        True,
+                        project_uuid,
+                    ),
+                    (area_entries, False, None),
+                    (extra_walls, False, None),
+                ],
             )
-            area_entries: list[tuple[float, dict[str, Any]]] = []
-            for row in rows if isinstance(rows, list) else []:
-                patch: dict[str, Any] = {
-                    key: float(row[key])
-                    for key in _AREA_KEYS
-                    if isinstance(row.get(key), (int, float))
-                }
-                if patch:
-                    area_entries.append((float(row["chainage_m"]), patch))
-            updated = await merge_cross_section_designs(
-                connection, route_id=route_id, entries=area_entries, replace=False
-            )
-            if extra_walls:
-                await merge_cross_section_designs(
-                    connection, route_id=route_id, entries=extra_walls, replace=False
-                )
             if isinstance(mass_haul, dict):
                 await merge_longitudinal_section_data(
                     connection, route_id=route_id, data_patch={"mass_haul": mass_haul}
@@ -351,7 +380,7 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int:
     marks.append(("정본 저장", time.perf_counter()))
     _log_steps("서버 재계산 내부", marks)
     logger.info(
-        "서버 재계산: route_id=%s 설계 보정 %s곳, 구조물 면적 %s곳, 유토곡선 %s",
+        "서버 재계산: route_id=%s 설계 보정 %s곳, 바뀐 행 %s곳, 유토곡선 %s",
         route_id,
         len(fixed),
         updated,

resources/tester/test_342_2_confirm_fingerprint.py

@@ -0,0 +1,273 @@
+# -*- coding: utf-8 -*-
+"""342-2 [확정] · [저장] 서버 재계산 — 입력 지문 · 바뀐 행만 · 행 읽기 한 번(부록 4 R23).
+
+① 지문 = Node 에 넘기는 입력 **전부** + 번들 판 — 같은 입력이면 Node 를 다시 안 돌림 ·
+   입력 어느 칸(깊은 칸 · 문맥 · 번들)이 바뀌어도 다시 돌림.
+② Node 에 가는 입력과 지문이 같은 한 벌 — 지문 밖으로 새는 입력 길이 없음.
+③ 행 쓰기 = 읽기 한 번 · 바뀐 행만 한 문장 · 앞 벌이 만든 새 행을 뒤 벌이 찾음.
+④ 같은 입력으로 두 번 = 두 번째는 Node 0 번 · 행 쓰기 0 · 유토곡선 쓰기 0.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+import B06_Section.B06_Section_Router as b06_router  # noqa: E402
+import B06_Section.B06_Section_Router_Design as b06_design  # noqa: E402
+import B06_Section.B06_Section_Server_Calc_Prebuild as prebuild  # noqa: E402
+from B06_Section.B06_Section_Repository_Bulk import (  # noqa: E402
+    merge_cross_section_design_batches,
+)
+
+DETAIL = {
+    "longitudinal": {"length_m": 40.0, "stations": [{"chainage_m": 0.0}]},
+    "cross_sections": [
+        {"chainage_m": 0.0, "design": {"ground_type": "soil", "slope": {"cut": 1.0}}},
+        {"chainage_m": 20.0, "design": {"ground_type": "rock"}},
+    ],
+}
+CONTEXT = {"earthwork_conversion": {"soil": {"L": 1.25, "C": 0.9}}, "haul_equipment_limits": []}
+
+
+@pytest.fixture
+def node(monkeypatch, tmp_path):
+    bundle = tmp_path / "bundle.js"
+    bundle.write_text("//", encoding="utf-8")
+    calls: list = []
+
+    def fake_run(path, script, payload):
+        calls.append(payload)
+        return {"areas": [], "mass_haul": {"total": len(calls)}}
+
+    monkeypatch.setattr(prebuild, "BUNDLE", bundle)
+    monkeypatch.setattr(prebuild, "run_bundle_json", fake_run)
+    monkeypatch.setattr(prebuild, "bundle_stale", lambda path: False)
+    monkeypatch.setattr(prebuild, "_NODE_CACHE", {})
+    return SimpleNamespace(calls=calls, bundle=bundle)
+
+
+def _input():
+    return {"detail": copy.deepcopy(DETAIL), "context": copy.deepcopy(CONTEXT)}
+
+
+def test_같은_입력이면_Node_한_번(node):
+    first = prebuild._node_output(_input())
+    second = prebuild._node_output(_input())
+    assert len(node.calls) == 1
+    assert first == second
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        lambda data: data["detail"]["cross_sections"][0]["design"]["slope"].update(cut=1.5),
+        lambda data: data["detail"]["cross_sections"].append({"chainage_m": 40.0}),
+        lambda data: data["detail"]["longitudinal"].update(length_m=40.5),
+        lambda data: data["context"]["earthwork_conversion"]["soil"].update(C=0.85),
+        lambda data: data.update(context={**data["context"], "structure_spoil_m3": 3.0}),
+    ],
+    ids=["깊은 설계 칸", "측점 더함", "종단", "환산계수", "문맥 새 칸"],
+)
+def test_입력_어느_칸이_바뀌어도_다시(node, change):
+    prebuild._node_output(_input())
+    changed = _input()
+    change(changed)
+    prebuild._node_output(changed)
+    assert len(node.calls) == 2
+
+
+def test_번들이_바뀌면_다시(node):
+    prebuild._node_output(_input())
+    stat = node.bundle.stat()
+    os.utime(node.bundle, ns=(stat.st_atime_ns, stat.st_mtime_ns + 10_000_000))
+    prebuild._node_output(_input())
+    assert len(node.calls) == 2
+
+
+def test_TS_가_번들보다_새로우면_다시(node, monkeypatch):
+    prebuild._node_output(_input())
+    monkeypatch.setattr(prebuild, "bundle_stale", lambda path: True)
+    prebuild._node_output(_input())
+    assert len(node.calls) == 2
+
+
+def test_실패는_기억_안_함(node, monkeypatch):
+    monkeypatch.setattr(prebuild, "run_bundle_json", lambda *a: node.calls.append(a) or None)
+    assert prebuild._node_output(_input()) is None
+    assert prebuild._node_output(_input()) is None
+    assert len(node.calls) == 2
+
+
+# ── ③ 행 쓰기 ──────────────────────────────────────────────────────────────
+
+
+class _Cursor:
+    def __init__(self, log, rows):
+        self.log, self.rows = log, rows
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def execute(self, sql, params=None):
+        self.log.append((" ".join(sql.split()), list(params or [])))
+
+    async def fetchall(self):
+        return self.rows
+
+
+class _Connection:
+    def __init__(self, rows):
+        self.log: list = []
+        self.rows = rows
+
+    def cursor(self, *args):
+        return _Cursor(self.log, self.rows)
+
+    async def begin(self):
+        return None
+
+    async def commit(self):
+        return None
+
+    async def rollback(self):
+        return None
+
+
+def _rows():
+    return [
+        (1, 0.0, {"design": {"ground_type": "soil", "cut_area_m2": 2.0}}),
+        (2, 20.0, json.dumps({"design": {"ground_type": "rock"}})),
+    ]
+
+
+def _kinds(log):
+    return [sql.split()[0] for sql, _ in log]
+
+
+def test_읽기_한_번_바뀐_행만(tmp_path):
+    connection = _Connection(_rows())
+    written = asyncio.run(
+        merge_cross_section_design_batches(
+            connection,
+            route_id=7,
+            batches=[
+                ([(0.0, {"ground_type": "soil", "cut_area_m2": 2.0})], True, None),  # 그대로
+                ([(20.0, {"cut_area_m2": 3.0})], False, None),  # 바뀜
+                ([(40.0, {"ground_type": "soil"})], True, uuid4()),  # 새 행
+                ([(40.0, {"cut_area_m2": 1.0})], False, None),  # 앞 벌의 새 행에 얹음
+            ],
+        )
+    )
+    assert _kinds(connection.log) == ["SELECT", "UPDATE", "INSERT"]
+    update = connection.log[1][1]
+    assert update[0] == 2 and json.loads(update[1])["design"] == {
+        "ground_type": "rock",
+        "cut_area_m2": 3.0,
+    }
+    insert = connection.log[2][1]
+    assert json.loads(insert[3]) == {"design": {"ground_type": "soil", "cut_area_m2": 1.0}}
+    assert written == 2
+
+
+def test_모두_그대로면_쓰기_없음():
+    connection = _Connection(_rows())
+    written = asyncio.run(
+        merge_cross_section_design_batches(
+            connection,
+            route_id=7,
+            batches=[
+                ([(0.0, {"cut_area_m2": 2.0})], False, None),
+                ([(20.0, {"ground_type": "rock"})], True, None),
+            ],
+        )
+    )
+    assert _kinds(connection.log) == ["SELECT"]
+    assert written == 0
+
+
+# ── ②④ 재계산 두 번 ──────────────────────────────────────────────────────────
+
+
+class _Pool:
+    def __init__(self, connection):
+        self.connection = connection
+
+    def acquire(self):
+        connection = self.connection
+
+        class _Acquire:
+            async def __aenter__(self):
+                return connection
+
+            async def __aexit__(self, *exc):
+                return False
+
+        return _Acquire()
+
+
+def test_같은_입력_두_번이면_Node_0_쓰기_0(node, monkeypatch, tmp_path):
+    rows = [(1, 0.0, {"design": DETAIL["cross_sections"][0]["design"]})]
+    connection = _Connection(rows)
+    stored = {"data": {"mass_haul": None}}
+    written_haul: list = []
+
+    async def detail(project_id, route_id):
+        return SimpleNamespace(model_dump=lambda mode: copy.deepcopy(DETAIL))
+
+    async def run_with(fn, *args):
+        if fn is prebuild.get_longitudinal_section:
+            return copy.deepcopy(stored)
+        if fn is prebuild.get_project_storage_relative_path:
+            return "p"
+        return {"source_filter": "csf", "method": "dtm", "smooth": False}
+
+    async def write_haul(connection, *, route_id, data_patch):
+        written_haul.append(data_patch)
+        stored["data"]["mass_haul"] = data_patch["mass_haul"]
+
+    def fake_run(path, script, payload):
+        node.calls.append(payload)
+        # Node 에 간 입력이 곧 지문 입력 — 다른 입력 길 없음
+        assert set(payload) == {"detail", "context"}
+        return {"areas": [{"chainage_m": 0.0, "cut_area_m2": 5.0}], "mass_haul": {"total": 9.0}}
+
+    monkeypatch.setattr(prebuild, "run_bundle_json", fake_run)
+    monkeypatch.setattr(b06_router, "get_section_detail", detail)
+    monkeypatch.setattr(prebuild, "run_with_connection", run_with)
+    monkeypatch.setattr(prebuild, "resolve_stored_project_path", lambda path: str(tmp_path))
+    monkeypatch.setattr(prebuild, "get_db_pool", lambda: _Pool(connection))
+    monkeypatch.setattr(prebuild, "merge_longitudinal_section_data", write_haul)
+    monkeypatch.setattr(prebuild, "quantity_settings", lambda root: {})
+    monkeypatch.setattr(prebuild, "mixed_conversion_factors", lambda settings: CONTEXT)
+    monkeypatch.setattr(prebuild, "haul_equipment_limits", lambda settings: [])
+    monkeypatch.setattr(b06_design, "enforce_design_lines", lambda *a: None)
+    monkeypatch.setattr(prebuild, "_enforce_stored_designs", lambda *a: None)
+
+    asyncio.run(prebuild.recompute_server_side(uuid4(), 3))
+    assert len(node.calls) == 1
+    assert _kinds(connection.log) == ["SELECT", "UPDATE"]
+    assert len(written_haul) == 1
+    # 첫 번째가 쓴 값을 행에 반영(가짜 DB) — 두 번째는 같은 입력
+    rows[0] = (1, 0.0, json.loads(connection.log[1][1][1]))
+    connection.log.clear()
+
+    asyncio.run(prebuild.recompute_server_side(uuid4(), 3))
+    assert len(node.calls) == 1  # Node 다시 안 돎
+    assert _kinds(connection.log) == ["SELECT"]  # 바뀐 행 없음 — 쓰기 0
+    assert len(written_haul) == 1  # 유토곡선 같은 값 — 안 씀