토큰 분산락 대기와 primary 조회에 타임아웃 적용

- readFromPrimary 트랜잭션에 타임아웃을 줘 발급 중인 노드의 락을 무한정 기다리지 않게 함
- getOutboundAccessToken 락 안 재확인도 readFromPrimary 로 읽어 near 캐시 옛 값으로 중복 발급하지 않게 함
- 락 대기/트랜잭션 타임아웃을 RMIInfo 설정으로 분리 (ms, 기본 15000/10000/3000)
  outboundTokenLockWaitMillis, outboundTokenReissueLockWaitMillis, outboundTokenReadTxTimeoutMillis
- 트랜잭션 타임아웃은 예상된 경우라 WARN 으로 기록

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 {
     
 
-	/** 아웃바운드 토큰 분산락 대기 시간(초). 스케줄러 경로에서 사용한다. */
+	/** 아웃바운드 토큰 분산락 대기 시간(ms) 설정 키 (RMIInfo 그룹) */
-	private static final int TOKEN_LOCK_WAIT_SECONDS = 60;
+	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<String, String> 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()) {

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);
+        transactions = Mockito.mock(IgniteTransactions.class);
-        Mockito.when(transactions.txStart(any(TransactionConcurrency.class), any(TransactionIsolation.class)))
+        Mockito.when(transactions.txStart(any(TransactionConcurrency.class), any(TransactionIsolation.class),
-                .thenReturn(transaction);
+                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());
+    }
 }