Initialize RTGS prototype repository

3센터 A-A-A RTGS 실시간총액결제 프로토타입 최초 버전관리 시작. - backend: Kotlin/Gradle 멀티모듈(sequencer, common, 채널/센터 모듈) - frontend: Vite + TS - infra: docker-compose, prometheus/grafana, ELK 네이티브 스크립트 - loadtest(k6), sim(장애/순서/정합성 시나리오), docs(PoC 보고서/복원력방안)
2026-07-15 15:37:32 +09:00
commit 58ca23b5d9
140 changed files with 11832 additions and 0 deletions
--- a/backend/hermes/build.gradle.kts
+++ b/backend/hermes/build.gradle.kts
@@ -0,0 +1,20 @@
+plugins {
+    kotlin("jvm")
+    kotlin("plugin.spring")
+    id("org.springframework.boot")
+    id("io.spring.dependency-management")
+}
+
+dependencies {
+    implementation(project(":common"))
+    implementation("org.springframework.boot:spring-boot-starter")
+    implementation("org.springframework.boot:spring-boot-starter-actuator")
+    implementation("org.springframework.boot:spring-boot-starter-web") // 메트릭/헬스 노출(B4)
+    implementation("org.springframework.boot:spring-boot-starter-data-jpa")
+    implementation("org.springframework.kafka:spring-kafka")
+    implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
+    implementation("org.jetbrains.kotlin:kotlin-reflect")
+    runtimeOnly("org.postgresql:postgresql")
+    testImplementation("org.springframework.boot:spring-boot-starter-test")
+    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
+}
--- a/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/GapBuffer.kt
+++ b/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/GapBuffer.kt
@@ -0,0 +1,67 @@
+package kr.or.bok.rtgs.hermes
+
+import kr.or.bok.rtgs.common.JournalEntry
+import java.util.concurrent.ConcurrentSkipListMap
+
+/**
+ * 저널 순차 게이트(B3) — globalSeq **오름차순**으로만 적용되도록 버퍼링/재정렬한다.
+ *
+ * rtgs.journal은 **단일 파티션**이라 Kafka가 offset(=순번기 발행) 순서대로 전달한다. 따라서
+ * "더 큰 seq가 도착했는데 `expected`는 안 온" 상황 = 그 `expected`가 **영영 발행되지 않은
+ * phantom**(B1의 SEQUENCE nextval 소비 후 미기록 등)이라는 의미다. 단일 파티션 순서 보장상
+ * 뒤늦게 채워질 일이 없으므로, 타임아웃 후 skip은 **실엔트리 스킵이 아니라** 결번 정리로 안전하다.
+ *
+ * 순수 로직(스프링 비의존) — 단위테스트로 규율 고정. Hermes 리스너(concurrency=1)와 타임아웃
+ * 스케줄러가 함께 접근하므로 메서드는 @Synchronized.
+ */
+class GapBuffer {
+    @Volatile
+    var expected: Long = 0L
+        private set
+
+    @Volatile
+    var lastAdvanceMillis: Long = 0L
+        private set
+
+    private val pending = ConcurrentSkipListMap<Long, JournalEntry>()
+
+    fun pendingCount(): Int = pending.size
+    fun highestPending(): Long? = if (pending.isEmpty()) null else pending.lastKey()
+
+    /** 도착 엔트리를 넣고, 지금 적용 가능한 엔트리들을 **순서대로** 반환(없으면 빈 리스트). */
+    @Synchronized
+    fun offer(entry: JournalEntry, nowMillis: Long): List<JournalEntry> {
+        if (expected == 0L) {          // 첫 엔트리가 기대 순번을 앵커링
+            expected = entry.globalSeq
+            lastAdvanceMillis = nowMillis
+        }
+        return when {
+            entry.globalSeq < expected -> emptyList()                         // old/dup
+            entry.globalSeq > expected -> { pending[entry.globalSeq] = entry; emptyList() } // gap → 버퍼
+            else -> drainFrom(entry, nowMillis)                               // 기대 순번 → 적용 + 이어받기
+        }
+    }
+
+    /** expected(=phantom 확정)를 건너뛰고, 이어서 적용 가능한 엔트리들을 반환. */
+    @Synchronized
+    fun skipExpected(nowMillis: Long): List<JournalEntry> {
+        if (pending.isEmpty()) return emptyList()   // 스킵할 이유 없음(대기자 없음)
+        expected++
+        val ready = mutableListOf<JournalEntry>()
+        while (pending.containsKey(expected)) { ready += pending.remove(expected)!!; expected++ }
+        if (ready.isNotEmpty()) lastAdvanceMillis = nowMillis
+        return ready
+    }
+
+    private fun drainFrom(head: JournalEntry, nowMillis: Long): List<JournalEntry> {
+        val ready = mutableListOf(head)
+        expected++
+        while (pending.containsKey(expected)) { ready += pending.remove(expected)!!; expected++ }
+        lastAdvanceMillis = nowMillis
+        return ready
+    }
+
+    /** pending이 있으면 마지막 진전 이후 경과(ms), 없으면 0(정체 아님). */
+    fun stalledMillis(nowMillis: Long): Long =
+        if (pending.isEmpty()) 0L else (nowMillis - lastAdvanceMillis).coerceAtLeast(0L)
+}
--- a/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/HermesApplication.kt
+++ b/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/HermesApplication.kt
@@ -0,0 +1,71 @@
+package kr.or.bok.rtgs.hermes
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import kr.or.bok.rtgs.common.ledger.Account
+import kr.or.bok.rtgs.common.ledger.JournalLog
+import kr.or.bok.rtgs.common.ledger.TransferRecord
+import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.autoconfigure.domain.EntityScan
+import org.springframework.boot.runApplication
+import org.springframework.context.annotation.Bean
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Modifying
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories
+import org.springframework.data.repository.query.Param
+import org.springframework.scheduling.annotation.EnableScheduling
+
+@SpringBootApplication
+@EntityScan("kr.or.bok.rtgs.common.ledger")
+@EnableJpaRepositories(considerNestedRepositories = true)
+@EnableScheduling  // B3 gap 타임아웃 스케줄러
+class HermesApplication {
+    @Bean
+    fun objectMapper(): ObjectMapper = jacksonObjectMapper()
+}
+
+interface AccountRepository : JpaRepository<Account, String>
+interface JournalLogRepository : JpaRepository<JournalLog, Long>
+
+interface TransferRepository : JpaRepository<TransferRecord, String> {
+    /**
+     * 거래 원장 원자적 upsert. 신규면 INSERT, 이미 있으면(Dior/타 소비자가 먼저 생성) UPDATE.
+     * created_at은 최초 값 유지(갱신 제외). Dior-Hermes 동시 삽입 경합에서 중복키 예외 제거.
+     */
+    @Modifying
+    @Query(
+        value = """
+            INSERT INTO transfer
+              (bmi, global_seq, status, sender_code, receiver_code, amount, origin_center, created_at, updated_at, reason)
+            VALUES
+              (:bmi, :seq, :status, :sender, :receiver, :amount, :origin, :createdAt, :updatedAt, :reason)
+            ON CONFLICT (bmi) DO UPDATE SET
+              global_seq    = EXCLUDED.global_seq,
+              status        = EXCLUDED.status,
+              sender_code   = EXCLUDED.sender_code,
+              receiver_code = EXCLUDED.receiver_code,
+              amount        = EXCLUDED.amount,
+              origin_center = EXCLUDED.origin_center,
+              updated_at    = EXCLUDED.updated_at,
+              reason        = EXCLUDED.reason
+        """,
+        nativeQuery = true,
+    )
+    fun upsertTransfer(
+        @Param("bmi") bmi: String,
+        @Param("seq") seq: Long,
+        @Param("status") status: String,
+        @Param("sender") sender: String,
+        @Param("receiver") receiver: String,
+        @Param("amount") amount: Long,
+        @Param("origin") origin: String,
+        @Param("createdAt") createdAt: Long,
+        @Param("updatedAt") updatedAt: Long,
+        @Param("reason") reason: String?,
+    ): Int
+}
+
+fun main(args: Array<String>) {
+    runApplication<HermesApplication>(*args)
+}
--- a/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/HermesService.kt
+++ b/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/HermesService.kt
@@ -0,0 +1,83 @@
+package kr.or.bok.rtgs.hermes
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.micrometer.core.instrument.MeterRegistry
+import io.micrometer.core.instrument.Timer
+import kr.or.bok.rtgs.common.JournalEntry
+import kr.or.bok.rtgs.common.ResultMessage
+import kr.or.bok.rtgs.common.Topics
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Service
+
+/**
+ * 저널 순차 소비 + 순서 탐지·교정(복원력방안 2.3-부속, S4·B3). 실제 정산은 SettlementApplier(트랜잭션).
+ * **결과(rtgs.result) 발행은 정산 트랜잭션 커밋 이후**에 한다(publish-after-commit) — Prada가
+ * 커밋 전 transfer를 읽어 최종확정을 놓치는 레이스 방지. rtgs.result는 B2에서 applied-ack으로도 쓰인다.
+ *
+ * 순서 게이트는 GapBuffer(단위테스트로 규율 고정). 단일 파티션이라 정상 시 결번이 없고, 결번은
+ * 곧 phantom(발행 안 된 seq)이므로 **타임아웃 후 skip**으로 정체를 자가해소한다(B3).
+ */
+@Service
+class HermesService(
+    private val applier: SettlementApplier,
+    private val kafka: KafkaTemplate<String, String>,
+    private val mapper: ObjectMapper,
+    private val metrics: MeterRegistry,
+    @Value("\${rtgs.gap-timeout-ms:5000}") private val gapTimeoutMs: Long,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    private val gate = GapBuffer()
+    // B4 메트릭: 정산 처리시간(TPS·지연) + gap 버퍼링 + gap 타임아웃(phantom skip)
+    private val settleTimer: Timer = Timer.builder("rtgs.settle").description("정산 처리시간").register(metrics)
+    private val gapCounter = metrics.counter("rtgs.journal.gap")
+    private val gapTimeoutCounter = metrics.counter("rtgs.journal.gap.timeout")
+
+    // 센터별 그룹(hermes-DC1/DC2/DC3)이라 각 센터가 저널 전량을 독립 재생.
+    @KafkaListener(topics = [Topics.JOURNAL], groupId = "hermes-\${rtgs.center-id:DC1}", concurrency = "1")
+    fun onJournal(payload: String) {
+        val entry = mapper.readValue(payload, JournalEntry::class.java)
+        val before = gate.pendingCount()
+        val ready = gate.offer(entry, System.currentTimeMillis())
+        if (ready.isEmpty() && gate.pendingCount() > before) {
+            gapCounter.increment()
+            log.warn("GAP: got seq={} expected={} -> buffered (pending={})", entry.globalSeq, gate.expected, gate.pendingCount())
+        }
+        ready.forEach { apply(it) }
+    }
+
+    /**
+     * B3 정체 감지: pending이 있고 마지막 진전 이후 gapTimeoutMs 초과면, expected는 phantom으로
+     * 확정(단일 파티션 순서 보장)하고 skip해 진행을 재개한다. 실엔트리 스킵이 아니라 결번 정리.
+     */
+    @Scheduled(fixedDelayString = "\${rtgs.gap-check-ms:1000}")
+    fun checkGapTimeout() {
+        val now = System.currentTimeMillis()
+        if (gate.stalledMillis(now) < gapTimeoutMs) return
+        val missing = gate.expected
+        val ready = gate.skipExpected(now)
+        if (ready.isNotEmpty()) {
+            gapTimeoutCounter.increment()
+            log.error("GAP TIMEOUT: seq={} 미도착 {}ms 초과 → phantom 확정 skip, 재개(적용 {}건, 다음 expected={})",
+                missing, gapTimeoutMs, ready.size, gate.expected)
+            ready.forEach { apply(it) }
+        }
+    }
+
+    /** 정산 적용(시간 계측) 후 결과 발행. */
+    private fun apply(entry: JournalEntry) {
+        val result = settleTimer.recordCallable { applier.apply(entry) }
+        publishAfterCommit(result)
+        if (result != null) metrics.counter("rtgs.settlements", "status", result.status.name).increment()
+    }
+
+    /** 트랜잭션 커밋 후(apply 반환 = 커밋 완료) 결과 발행. null이면 멱등 스킵. */
+    private fun publishAfterCommit(result: ResultMessage?) {
+        if (result == null) return
+        kafka.send(Topics.RESULT, result.journalEntry.core.bmi, mapper.writeValueAsString(result))
+    }
+}
--- a/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/SettlementApplier.kt
+++ b/backend/hermes/src/main/kotlin/kr/or/bok/rtgs/hermes/SettlementApplier.kt
@@ -0,0 +1,74 @@
+package kr.or.bok.rtgs.hermes
+
+import kr.or.bok.rtgs.common.JournalEntry
+import kr.or.bok.rtgs.common.ResultMessage
+import kr.or.bok.rtgs.common.TxSts
+import kr.or.bok.rtgs.common.ledger.JournalLog
+import kr.or.bok.rtgs.common.ledger.TransferRecord
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+
+/**
+ * 실제 정산을 강한 일관성 트랜잭션으로 수행. 한 저널 엔트리 = 한 트랜잭션.
+ * 선저널 → 잔액 이체 → transfer 상태확정. **결과 발행은 하지 않고 ResultMessage만 반환**한다
+ * (Kafka 발행은 호출측 HermesService가 트랜잭션 커밋 이후에 수행 — publish-after-commit).
+ * 이렇게 해야 Prada가 커밋 전 transfer를 읽는 레이스(ACSP 잔류)를 방지한다.
+ */
+@Service
+class SettlementApplier(
+    private val accounts: AccountRepository,
+    private val transfers: TransferRepository,
+    private val journalLogs: JournalLogRepository,
+    @Value("\${rtgs.center-id:DC1}") private val centerId: String,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    /** @return 발행할 ResultMessage. 멱등 스킵이면 null. */
+    @Transactional
+    fun apply(e: JournalEntry): ResultMessage? {
+        val bmi = e.core.bmi
+        val existing = transfers.findById(bmi).orElse(null)
+        if (existing != null && (existing.status == TxSts.ACSP || existing.status == TxSts.ACCC)) return null // 멱등
+
+        if (!journalLogs.existsById(e.globalSeq)) {
+            journalLogs.save(JournalLog(globalSeq = e.globalSeq, bmi = bmi, seqEpochMillis = e.seqEpochMillis, applied = false))
+        }
+
+        val now = System.currentTimeMillis()
+        val sender = accounts.findById(e.core.senderCode).orElse(null)
+        val receiver = accounts.findById(e.core.receiverCode).orElse(null)
+
+        if (sender == null || receiver == null) {
+            return settle(e, TxSts.RJCT, now, null, null, "unknown account(s): ${e.core.senderCode}/${e.core.receiverCode}")
+        }
+        if (sender.balance < e.core.amount) {
+            return settle(e, TxSts.RJCT, now, sender.balance, receiver.balance, "insufficient balance")
+        }
+
+        // 이체(강한 일관성): 송신기관 당좌 차감 + 수신기관 당좌 가산
+        sender.balance -= e.core.amount
+        receiver.balance += e.core.amount
+        accounts.save(sender)
+        accounts.save(receiver)
+        val result = settle(e, TxSts.ACSP, now, sender.balance, receiver.balance, null)
+        log.info("ACSP seq=#{} bmi={} {}({})->{}({})", e.globalSeq, bmi,
+            e.core.senderCode, sender.balance, e.core.receiverCode, receiver.balance)
+        return result
+    }
+
+    private fun settle(e: JournalEntry, status: TxSts, now: Long, dBal: Long?, cBal: Long?, reason: String?): ResultMessage {
+        // 원자적 upsert(신규 INSERT / 기존 UPDATE) — Dior와의 동시 삽입 경합에서 중복키 예외 없음.
+        transfers.upsertTransfer(
+            bmi = e.core.bmi, seq = e.globalSeq, status = status.name,
+            sender = e.core.senderCode, receiver = e.core.receiverCode, amount = e.core.amount,
+            origin = e.originCenter, createdAt = e.seqEpochMillis, updatedAt = now, reason = reason,
+        )
+        journalLogs.findById(e.globalSeq).ifPresent { it.applied = true; journalLogs.save(it) }
+        return ResultMessage(
+            journalEntry = e, status = status, processedCenter = centerId,
+            debtorBalanceAfter = dBal, creditorBalanceAfter = cBal, reason = reason,
+        )
+    }
+}
--- a/backend/hermes/src/main/resources/application.yml
+++ b/backend/hermes/src/main/resources/application.yml
@@ -0,0 +1,37 @@
+spring:
+  application:
+    name: hermes
+  datasource:
+    url: jdbc:postgresql://${POSTGRES_HOST:localhost}:${POSTGRES_PORT:5433}/${POSTGRES_DB:rtgs}
+    username: ${POSTGRES_USER:rtgs}
+    password: ${POSTGRES_PASSWORD:rtgs}
+  jpa:
+    hibernate:
+      ddl-auto: none
+    open-in-view: false
+  kafka:
+    bootstrap-servers: ${KAFKA_BOOTSTRAP:localhost:9092}
+    consumer:
+      group-id: hermes
+      auto-offset-reset: earliest
+      enable-auto-commit: false
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
+      acks: all
+    listener:
+      ack-mode: record
+
+server:
+  port: 8093
+
+rtgs:
+  center-id: ${CENTER_ID:DC1}
+
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,info,prometheus,metrics
--- a/backend/hermes/src/test/kotlin/kr/or/bok/rtgs/hermes/GapBufferTest.kt
+++ b/backend/hermes/src/test/kotlin/kr/or/bok/rtgs/hermes/GapBufferTest.kt
@@ -0,0 +1,80 @@
+package kr.or.bok.rtgs.hermes
+
+import kr.or.bok.rtgs.common.CoreMessage
+import kr.or.bok.rtgs.common.JournalEntry
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+/** B3 순차 게이트(버퍼링·재정렬·phantom skip) 규율 고정 테스트. */
+class GapBufferTest {
+
+    private fun entry(seq: Long): JournalEntry = JournalEntry(
+        globalSeq = seq,
+        seqEpochMillis = 1_000L + seq,
+        originCenter = "DC1",
+        core = CoreMessage(
+            bmi = "bmi%022d".format(seq), msgType = "pacs.008.001.08",
+            senderCode = "1001", receiverCode = "1002",
+            debtorAcct = "1001", creditorAcct = "1002",
+            amount = 100, origHash = "h",
+        ),
+    )
+
+    private fun seqs(list: List<JournalEntry>) = list.map { it.globalSeq }
+
+    @Test
+    fun `순서대로 오면 즉시 적용`() {
+        val g = GapBuffer()
+        assertEquals(listOf(1L), seqs(g.offer(entry(1), 10)))
+        assertEquals(listOf(2L), seqs(g.offer(entry(2), 11)))
+        assertEquals(3L, g.expected)
+    }
+
+    @Test
+    fun `역전 도착은 버퍼링 후 결번 채워지면 한꺼번에 방출`() {
+        val g = GapBuffer()
+        assertEquals(listOf(1L), seqs(g.offer(entry(1), 10)))
+        assertEquals(emptyList<Long>(), seqs(g.offer(entry(3), 11))) // gap: 2 미도착 → 버퍼
+        assertEquals(emptyList<Long>(), seqs(g.offer(entry(4), 12))) // 버퍼
+        assertEquals(2, g.pendingCount())
+        assertEquals(listOf(2L, 3L, 4L), seqs(g.offer(entry(2), 13))) // 2 도착 → 2,3,4 연속 방출
+        assertEquals(5L, g.expected)
+        assertEquals(0, g.pendingCount())
+    }
+
+    @Test
+    fun `과거 중복 seq는 무시`() {
+        val g = GapBuffer()
+        g.offer(entry(1), 10); g.offer(entry(2), 11)
+        assertEquals(emptyList<Long>(), seqs(g.offer(entry(1), 12))) // 이미 지난 seq
+        assertEquals(3L, g.expected)
+    }
+
+    @Test
+    fun `첫 엔트리가 중간 seq여도 그 값으로 앵커링`() {
+        val g = GapBuffer()
+        assertEquals(listOf(5L), seqs(g.offer(entry(5), 10))) // 재기동 후 earliest=5
+        assertEquals(6L, g.expected)
+    }
+
+    @Test
+    fun `정체 감지 phantom skip으로 진행 재개`() {
+        val g = GapBuffer()
+        g.offer(entry(1), 100)                 // 적용, expected=2, lastAdvance=100
+        g.offer(entry(3), 101)                 // 2 결번 → 3 버퍼
+        assertEquals(50L, g.stalledMillis(150)) // 150-100 정체
+        // 타임아웃: 2는 phantom → skip → 3 방출, expected=4
+        assertEquals(listOf(3L), seqs(g.skipExpected(160)))
+        assertEquals(4L, g.expected)
+        assertEquals(0, g.stalledMillis(200))   // 대기자 없음 → 정체 아님
+    }
+
+    @Test
+    fun `대기자 없으면 정체 0 이고 skip은 무동작`() {
+        val g = GapBuffer()
+        g.offer(entry(1), 10)
+        assertEquals(0L, g.stalledMillis(9999))
+        assertEquals(emptyList<Long>(), seqs(g.skipExpected(9999)))
+        assertEquals(2L, g.expected) // 변화 없음
+    }
+}