From ac5e617b37721147ae884f7c13daf0b94e0a0b74 Mon Sep 17 00:00:00 2001 From: curry772 Date: Mon, 14 Sep 2026 10:52:03 +0900 Subject: [PATCH] =?UTF-8?q?=ED=86=A0=ED=81=B0=20=EB=B6=84=EC=82=B0?= =?UTF-8?q?=EB=9D=BD=20=EB=8C=80=EA=B8=B0=EC=99=80=20primary=20=EC=A1=B0?= =?UTF-8?q?=ED=9A=8C=EC=97=90=20=ED=83=80=EC=9E=84=EC=95=84=EC=9B=83=20?= =?UTF-8?q?=EC=A0=81=EC=9A=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - readFromPrimary 트랜잭션에 타임아웃을 줘 발급 중인 노드의 락을 무한정 기다리지 않게 함 - getOutboundAccessToken 락 안 재확인도 readFromPrimary 로 읽어 near 캐시 옛 값으로 중복 발급하지 않게 함 - 락 대기/트랜잭션 타임아웃을 RMIInfo 설정으로 분리 (ms, 기본 15000/10000/3000) outboundTokenLockWaitMillis, outboundTokenReissueLockWaitMillis, outboundTokenReadTxTimeoutMillis - 트랜잭션 타임아웃은 예상된 경우라 WARN 으로 기록 --- .../session/SessionManagerForIgnite.java | 74 ++++++++++++++++--- .../SessionManagerForIgniteReissueTest.java | 53 ++++++++++++- 2 files changed, 113 insertions(+), 14 deletions(-) diff --git a/src/main/java/com/eactive/eai/common/session/SessionManagerForIgnite.java b/src/main/java/com/eactive/eai/common/session/SessionManagerForIgnite.java index c0cac5f..9a1c20d 100644 --- a/src/main/java/com/eactive/eai/common/session/SessionManagerForIgnite.java +++ b/src/main/java/com/eactive/eai/common/session/SessionManagerForIgnite.java @@ -77,11 +77,29 @@ import ch.qos.logback.classic.Level; public class SessionManagerForIgnite extends SessionManager { - /** 아웃바운드 토큰 분산락 대기 시간(초). 스케줄러 경로에서 사용한다. */ - private static final int TOKEN_LOCK_WAIT_SECONDS = 60; + /** 아웃바운드 토큰 분산락 대기 시간(ms) 설정 키 (RMIInfo 그룹) */ + public static final String OUTBOUND_TOKEN_LOCK_WAIT_MILLIS = "outboundTokenLockWaitMillis"; + /** 거래 중 재발급(기관 거부) 분산락 대기 시간(ms) 설정 키 (RMIInfo 그룹) */ + public static final String OUTBOUND_TOKEN_REISSUE_LOCK_WAIT_MILLIS = "outboundTokenReissueLockWaitMillis"; + /** 아웃바운드 토큰 primary 읽기 트랜잭션 타임아웃(ms) 설정 키 (RMIInfo 그룹) */ + public static final String OUTBOUND_TOKEN_READ_TX_TIMEOUT_MILLIS = "outboundTokenReadTxTimeoutMillis"; - /** 거래 중 재발급은 거래 스레드를 잡으므로 스케줄러 경로보다 짧게 기다린다. */ - private static final int REISSUE_LOCK_WAIT_SECONDS = 10; + /** + * 아웃바운드 토큰 분산락 대기 시간(ms). RMIInfo.outboundTokenLockWaitMillis + * 토큰이 없거나 만료일 때 거래 스레드도 여기서 기다리므로 길게 잡지 않는다. + * (스케줄러는 락을 못 잡으면 다른 노드가 발급 중이라는 뜻이라 오래 기다릴 필요가 없다) + */ + private long tokenLockWaitMillis = 15000L; + + /** 거래 중 재발급(기관 거부) 분산락 대기 시간(ms). RMIInfo.outboundTokenReissueLockWaitMillis */ + private long reissueLockWaitMillis = 10000L; + + /** + * readFromPrimary 트랜잭션 타임아웃(ms). RMIInfo.outboundTokenReadTxTimeoutMillis + * 비관적 트랜잭션 읽기는 키 락을 잡으므로 다른 노드가 lock(key) 로 발급 중이면 발급이 끝날 때까지 기다린다. + * 타임아웃이 없으면 이 대기는 락 대기 시간(tryLock)의 제한을 받지 않는다. + */ + private long readTxTimeoutMillis = 3000L; // evictMaster Instance - single socket 관련 private static IgniteCache evictMasterCache = null; // Socket Session cache 정보 @@ -219,6 +237,15 @@ public class SessionManagerForIgnite extends SessionManager { cacheStoreMaxSize = 10000; } } + + tokenLockWaitMillis = parseMillis(prpty, OUTBOUND_TOKEN_LOCK_WAIT_MILLIS, tokenLockWaitMillis, 1000L); + reissueLockWaitMillis = parseMillis(prpty, OUTBOUND_TOKEN_REISSUE_LOCK_WAIT_MILLIS, reissueLockWaitMillis, 1000L); + readTxTimeoutMillis = parseMillis(prpty, OUTBOUND_TOKEN_READ_TX_TIMEOUT_MILLIS, readTxTimeoutMillis, 100L); + + if (logger.isWarn()) { + logger.warn(">> OutboundToken tokenLockWaitMillis = " + tokenLockWaitMillis + ", reissueLockWaitMillis = " + + reissueLockWaitMillis + ", readTxTimeoutMillis = " + readTxTimeoutMillis); + } IgniteConfiguration config = new IgniteConfiguration(); @@ -449,7 +476,28 @@ public class SessionManagerForIgnite extends SessionManager { cacheWebSocketTimeout = manager.cache(WEBSOCKER_CAHCE_NAME); cacheTerminal = manager.cache(TERMINAL_CAHCE_NAME); } - + + /** + * 밀리초 설정값을 읽는다. 없거나 숫자가 아니거나 최소값보다 작으면 기본값을 쓴다. + */ + private long parseMillis(Properties prpty, String key, long defaultValue, long minValue) { + String value = prpty == null ? null : prpty.getProperty(key); + if (StringUtils.isBlank(value)) { + return defaultValue; + } + try { + long millis = Long.parseLong(value.trim()); + if (millis < minValue) { + logger.warn(">> " + key + " = " + value + " is less than " + minValue + ". use default " + defaultValue); + return defaultValue; + } + return millis; + } catch (NumberFormatException ex) { + logger.warn(">> " + key + " = " + value + " is not a number. use default " + defaultValue); + return defaultValue; + } + } + private CacheConfiguration initCache(IgniteConfiguration config, int expiryDurationSecs, String bootstrapAsynchronously, String name, int maxsize, boolean useCacheWriteModeAsync, boolean transactionMode) { @@ -848,17 +896,20 @@ public class SessionManagerForIgnite extends SessionManager { * 비관적 트랜잭션 안에서 읽으면 primary 의 값을 보장받는다. 트랜잭션은 읽기 직후 바로 닫는다. * (발급 HTTP 호출 구간까지 열어두면 장기 트랜잭션이 되어 파티션 맵 교환을 막는다) * + * 다른 노드가 발급 중이라 키 락을 기다리게 되면 readTxTimeoutMillis 후 포기하고 일반 get() 으로 읽는다. + * * @param key 어댑터그룹명 * @return 캐시에 있는 토큰. 없으면 null. */ private AccessTokenVO readFromPrimary(String key) { try (Transaction tx = manager.transactions().txStart(TransactionConcurrency.PESSIMISTIC, - TransactionIsolation.REPEATABLE_READ)) { + TransactionIsolation.REPEATABLE_READ, readTxTimeoutMillis, 1)) { AccessTokenVO token = cacheOutBoundAccessToken.get(key); tx.commit(); return token; } catch (Throwable e) { - logger.error("occuring exception in readFromPrimary. key=" + key, e); + // 다른 노드가 발급 중이라 트랜잭션 타임아웃으로 빠지는 것은 예상된 경우라 WARN 으로 남긴다. + logger.warn("occuring exception in readFromPrimary. fallback to get(). key=" + key, e); return cacheOutBoundAccessToken.get(key); } } @@ -907,7 +958,7 @@ public class SessionManagerForIgnite extends SessionManager { boolean acquired = false; try { - acquired = lock.tryLock(REISSUE_LOCK_WAIT_SECONDS, TimeUnit.SECONDS); + acquired = lock.tryLock(reissueLockWaitMillis, TimeUnit.MILLISECONDS); if (!acquired) { logger.warn("reissueOutboundAccessToken lock timeout. key=" + key); @@ -1039,12 +1090,13 @@ public class SessionManagerForIgnite extends SessionManager { try { logger.debug("calling func getOutboundAccessToken = distributed ignite trylock"); - acquired = lock.tryLock(TOKEN_LOCK_WAIT_SECONDS, TimeUnit.SECONDS); + acquired = lock.tryLock(tokenLockWaitMillis, TimeUnit.MILLISECONDS); if ( acquired ) { logger.debug("calling func getOutboundAccessToken = distributed ignite trylock success=>"+acquired); // 3. 락 획득 후 다시 확인 (그 사이 다른 인스턴스가 갱신했을 수 있음) - token = cacheOutBoundAccessToken.get(key); + // near 캐시 사본은 무효화가 늦을 수 있어 get() 으로는 방금 발급된 토큰을 놓치고 다시 발급할 수 있다. + token = readFromPrimary(key); logger.debug(String.format("getting token in ignite=%s", token)); @@ -1094,7 +1146,7 @@ public class SessionManagerForIgnite extends SessionManager { try { logger.debug("calling func removeOutboundAccessToken = distributed ignite trylock"); - acquired = lock.tryLock(TOKEN_LOCK_WAIT_SECONDS, TimeUnit.SECONDS); + acquired = lock.tryLock(tokenLockWaitMillis, TimeUnit.MILLISECONDS); if( acquired ) { if (logger.isDebug()) { diff --git a/src/test/java/com/eactive/eai/common/session/SessionManagerForIgniteReissueTest.java b/src/test/java/com/eactive/eai/common/session/SessionManagerForIgniteReissueTest.java index 07409de..44c8388 100644 --- a/src/test/java/com/eactive/eai/common/session/SessionManagerForIgniteReissueTest.java +++ b/src/test/java/com/eactive/eai/common/session/SessionManagerForIgniteReissueTest.java @@ -58,6 +58,9 @@ class SessionManagerForIgniteReissueTest { /** readFromPrimary 가 여는 트랜잭션 */ private Transaction transaction; + /** readFromPrimary 가 트랜잭션을 여는 곳 (타임아웃 인자 검증용) */ + private IgniteTransactions transactions; + @BeforeEach void setUp() throws Exception { sessionManager = new SessionManagerForIgnite(); @@ -86,9 +89,9 @@ class SessionManagerForIgniteReissueTest { */ private void injectIgnite() throws Exception { transaction = Mockito.mock(Transaction.class); - IgniteTransactions transactions = Mockito.mock(IgniteTransactions.class); - Mockito.when(transactions.txStart(any(TransactionConcurrency.class), any(TransactionIsolation.class))) - .thenReturn(transaction); + transactions = Mockito.mock(IgniteTransactions.class); + Mockito.when(transactions.txStart(any(TransactionConcurrency.class), any(TransactionIsolation.class), + Mockito.anyLong(), Mockito.anyInt())).thenReturn(transaction); Ignite ignite = Mockito.mock(Ignite.class); Mockito.when(ignite.transactions()).thenReturn(transactions); @@ -316,4 +319,48 @@ class SessionManagerForIgniteReissueTest { assertTrue(locks.containsKey("GRP-A") && locks.containsKey("GRP-B"), "그룹별로 락이 분리돼야 한다"); assertSame(locks.get("GRP-A"), locks.get("GRP-A")); } + + @Test + @DisplayName("8. 트랜잭션 읽기에는 타임아웃을 준다 (발급 중인 노드의 락을 무한정 기다리지 않도록)") + void 트랜잭션읽기_타임아웃() { + store.put(GROUP, token("OLD")); + + sessionManager.reissueOutboundAccessToken(GROUP, "OLD", 0L, supplier(new AtomicInteger(), "NEW")); + + Mockito.verify(transactions, Mockito.times(2)).txStart( + Mockito.eq(TransactionConcurrency.PESSIMISTIC), Mockito.eq(TransactionIsolation.REPEATABLE_READ), + Mockito.longThat(timeout -> timeout > 0), Mockito.anyInt()); + } + + @Test + @DisplayName("9. getOutboundAccessToken 도 락 안의 재확인은 트랜잭션으로 읽는다") + void 최초발급_락안에서_트랜잭션읽기() { + AtomicInteger issued = new AtomicInteger(); + + AccessTokenVO result = sessionManager.getOutboundAccessToken(GROUP, supplier(issued, "NEW")); + + assertEquals(1, issued.get()); + assertEquals("NEW", result.getAccessToken()); + Mockito.verify(transaction, Mockito.times(1)).commit(); + } + + @Test + @DisplayName("10. getOutboundAccessToken - 락 안에서 읽은 토큰이 유효하면 발급하지 않는다") + void 최초발급_락안에서_이미발급됨() { + // 락 밖 get() 은 옛 만료 토큰(near 캐시 사본), 락 안의 트랜잭션 읽기는 다른 노드가 방금 넣은 토큰을 돌려준다. + OAuth2AccessTokenVO expired = token("OLD"); + expired.setExpiration(new Date(System.currentTimeMillis() - 1_000L)); + store.put(GROUP, expired); + Mockito.when(transactions.txStart(any(TransactionConcurrency.class), any(TransactionIsolation.class), + Mockito.anyLong(), Mockito.anyInt())).thenAnswer(invocation -> { + store.put(GROUP, token("ALREADY-NEW")); + return transaction; + }); + + AtomicInteger issued = new AtomicInteger(); + AccessTokenVO result = sessionManager.getOutboundAccessToken(GROUP, supplier(issued, "NEW")); + + assertEquals(0, issued.get(), "락 안에서 확인한 토큰이 유효하면 발급하지 않아야 한다"); + assertEquals("ALREADY-NEW", result.getAccessToken()); + } }