diff --git a/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisQueueTokenRepository.java b/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisQueueTokenRepository.java index 7ff0b3b..4cdcf3b 100644 --- a/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisQueueTokenRepository.java +++ b/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisQueueTokenRepository.java @@ -9,31 +9,38 @@ import com.firstticket.queueservice.queuetoken.domain.vo.ProgramId; import com.firstticket.queueservice.queuetoken.domain.vo.QueueTokenId; import com.firstticket.queueservice.queuetoken.domain.vo.UserId; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.core.*; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.data.redis.core.HashOperations; +import org.springframework.data.redis.core.RedisOperations; +import org.springframework.data.redis.core.SessionCallback; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Repository; -import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.util.*; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; /** * 대기 토큰의 Redis 기반 영속성 구현체. * - *

3가지 자료구조를 조합하여 사용한다: + *

5가지 자료구조를 조합하여 사용한다: *

* - *

저장 시 역인덱스(setIfAbsent)와 트랜잭션(MULTI/EXEC)으로 일관성을 보장한다. - * 자세한 흐름은 {@link #enqueue(QueueToken)} 참고. + *

enqueue / delete / deleteAllByProgramId 는 Lua 스크립트로 원자 처리한다.

*/ @Slf4j @Repository -@RequiredArgsConstructor public class RedisQueueTokenRepository implements QueueTokenRepository { // ===== 키 prefix 상수 ===== @@ -41,6 +48,8 @@ public class RedisQueueTokenRepository implements QueueTokenRepository { private static final String PROGRAM_KEY_PREFIX = "program:"; private static final String TOKEN_KEY_PREFIX = "token:"; private static final String USER_KEY_PREFIX = "user:"; + private static final String SEQ_KEY_PREFIX = "queue:seq:"; + private static final String PROGRAM_TOKENS_SUFFIX = ":tokens"; // ===== Hash 필드 이름 상수 ===== private static final String FIELD_USER_ID = "userId"; @@ -51,72 +60,62 @@ public class RedisQueueTokenRepository implements QueueTokenRepository { private final StringRedisTemplate redisTemplate; private final QueueProperties properties; + private final DefaultRedisScript enqueueScript; + private final DefaultRedisScript deleteScript; + private final DefaultRedisScript deleteAllByProgramScript; + + public RedisQueueTokenRepository( + StringRedisTemplate redisTemplate, + QueueProperties properties, + @Qualifier("enqueueScript") DefaultRedisScript enqueueScript, + @Qualifier("deleteScript") DefaultRedisScript deleteScript, + @Qualifier("deleteAllByProgramScript") DefaultRedisScript deleteAllByProgramScript + ) { + this.redisTemplate = redisTemplate; + this.properties = properties; + this.enqueueScript = enqueueScript; + this.deleteScript = deleteScript; + this.deleteAllByProgramScript = deleteAllByProgramScript; + } /** - * Redis 기반 enqueue 구현. + * Redis 기반 enqueue 구현 (Lua 스크립트로 원자 처리). * - *

2단계 처리: + *

Lua 스크립트 안에서 한 트랜잭션으로: *

    - *
  1. 역인덱스(setIfAbsent)를 락처럼 사용하여 중복 진입 방지
    2. - *
  2. Sorted Set + Hash 를 MULTI/EXEC 트랜잭션으로 저장
    3. + *
  3. 역인덱스 SETNX 로 중복 진입 방지
    4. + *
  4. INCR 시퀀스 + epoch_second 로 tie-breaker score 생성
    5. + *
  5. Sorted Set 추가 (ZADD)
    6. + *
  6. Hash 메타 저장 (HSET) + TTL (EXPIRE)
    7. + *
  7. 프로그램 단위 토큰 인덱스 (Set) 추가 (SADD)
    8. *
- * - *

한계: 1단계 성공 후 2단계 실패 시 orphan 역인덱스가 남을 수 있다. - * TTL 로 자동 정리되지만, 진짜 원자성을 위해 향후 Lua Script 도입 예정. - * - *

Sorted Set 의 멤버는 키 단위 TTL 불가하므로, - * {@link #delete(QueueToken)} 호출 시 명시적 ZREM 으로 정리한다. */ @Override public void enqueue(QueueToken token) { - - // 키 생성 String programKey = programKey(token.getProgramId()); String tokenKey = tokenKey(token.getId()); String userProgramKey = userProgramKey(token.getUserId(), token.getProgramId()); + String seqKey = seqKey(token.getProgramId()); + String programTokensKey = programTokensKey(token.getProgramId()); String tokenIdStr = token.getId().asString(); - - // Sorted Set의 score로 사용할 진입 시각 (epoch milli) long issuedAtEpochMilli = token.getIssuedAt().toEpochMilli(); + long ttlSeconds = properties.waitingTtl().getSeconds(); + + Long result = redisTemplate.execute( + enqueueScript, + List.of(userProgramKey, programKey, tokenKey, seqKey, programTokensKey), + tokenIdStr, + token.getUserId().asString(), + token.getProgramId().asString(), + String.valueOf(issuedAtEpochMilli), + token.getStatus().name(), + String.valueOf(ttlSeconds) + ); - Duration ttl = properties.waitingTtl(); - - // 1단계: 역인덱스를 락처럼 사용하여 중복 진입 방지 - Boolean acquired = redisTemplate.opsForValue() - .setIfAbsent(userProgramKey, tokenIdStr, ttl); - - if (!Boolean.TRUE.equals(acquired)) { + if (result == null || result == 0L) { throw new DuplicateTokenException(); } - - // 2단계: 나머지 키를 트랜잭션으로 저장 - Map tokenFields = Map.of( - FIELD_USER_ID, token.getUserId().asString(), - FIELD_PROGRAM_ID, token.getProgramId().asString(), - FIELD_ISSUED_AT, String.valueOf(issuedAtEpochMilli), - FIELD_STATUS, token.getStatus().name() - ); - - // 트랜잭션으로 Sorted Set + Hash 저장 + Hash TTL 설정 - redisTemplate.execute(new SessionCallback>() { - @SuppressWarnings({"unchecked"}) - @Override - public List execute(RedisOperations operations) { - operations.multi(); - - // 1. Sorted Set: 대기열에 추가 (score = 진입 시각) - operations.opsForZSet().add(programKey, tokenIdStr, issuedAtEpochMilli); - - // 2. Hash: 토큰 메타 데이터 저장 - operations.opsForHash().putAll(tokenKey, tokenFields); - - // 3. Hash 키에 TTL 설정 (Sorted Set은 컬렉션이라 별도 정리 필요) - operations.expire(tokenKey, ttl); - - return operations.exec(); - } - }); } /** @@ -177,7 +176,6 @@ public Optional findByUserIdAndProgramId(UserId userId, ProgramId pr // 1단계: 역인덱스로 tokenId 조회 String tokenIdStr = redisTemplate.opsForValue().get(userProgramKey); - if (tokenIdStr == null) { return Optional.empty(); } @@ -187,46 +185,31 @@ public Optional findByUserIdAndProgramId(UserId userId, ProgramId pr } /** - * Redis 기반 delete 구현. + * Redis 기반 delete 구현 (Lua 스크립트로 원자 처리). * - *

3개 키를 트랜잭션으로 묶어 삭제한다: + *

Lua 스크립트 안에서 한 트랜잭션으로: *

    - *
  1. Sorted Set 에서 토큰 멤버 제거 (ZREM)
    2. - *
  2. Hash 키 삭제
    3. - *
  3. 역인덱스 키 삭제
    4. + *
  4. 역인덱스 compare-and-delete (다른 토큰 차지 시 보존)
    5. + *
  5. Sorted Set 멤버 제거 (ZREM)
    6. + *
  6. Hash 키 삭제 (DEL)
    7. + *
  7. 프로그램 단위 토큰 인덱스에서 제거 (SREM)
    8. *
* - *

이미 만료/삭제된 토큰에 대해서도 안전하게 호출 가능 (멱등). + *

이미 만료 / 삭제된 토큰에 대해서도 안전하게 호출 가능 (멱등).

*/ @Override public void delete(QueueToken token) { String programKey = programKey(token.getProgramId()); String tokenKey = tokenKey(token.getId()); String userProgramKey = userProgramKey(token.getUserId(), token.getProgramId()); + String programTokensKey = programTokensKey(token.getProgramId()); String tokenIdStr = token.getId().asString(); - // 역인덱스 값과 토큰 ID가 일치할 때만 삭제 - // (다른 토큰이 차지한 경우는 그대로 둠 — race condition 방어) - String current = redisTemplate.opsForValue().get(userProgramKey); - if (tokenIdStr.equals(current)) { - redisTemplate.delete(userProgramKey); - } - - redisTemplate.execute(new SessionCallback>() { - @SuppressWarnings({"unchecked"}) - @Override - public List execute(RedisOperations operations) { - operations.multi(); - - // 1. Sorted Set 에서 멤버 제거 - operations.opsForZSet().remove(programKey, tokenIdStr); - - // 2. Hash 키 삭제 - operations.delete(tokenKey); - - return operations.exec(); - } - }); + redisTemplate.execute( + deleteScript, + List.of(userProgramKey, programKey, tokenKey, programTokensKey), + tokenIdStr + ); } /** @@ -246,7 +229,6 @@ public Optional findPosition(UserId userId, ProgramId programId) { // 1단계: 역인덱스로 tokenId 조회 String tokenIdStr = redisTemplate.opsForValue().get(userProgramKey); - if (tokenIdStr == null) { return Optional.empty(); } @@ -254,7 +236,6 @@ public Optional findPosition(UserId userId, ProgramId programId) { // 2단계: Sorted Set에서 순번 조회 (0-based) String programKey = programKey(programId); Long rank = redisTemplate.opsForZSet().rank(programKey, tokenIdStr); - if (rank == null) { return Optional.empty(); } @@ -266,14 +247,7 @@ public Optional findPosition(UserId userId, ProgramId programId) { /** * Redis 기반 findAdmissionCandidates 구현. * - *

2단계 조회: - *

    - *
  1. Sorted Set ZRANGE로 앞에서 batchSize개의 tokenId 추출
    2. - *
  2. 각 tokenId로 토큰 전체 조회 (N번 Hash 조회)
    3. - *
- * - *

Hash가 만료되어 토큰을 복원할 수 없는 orphan 케이스는 자연 필터링된다. - * 향후 N번 조회를 단일 Lua Script로 통합 검토. + *

Hash 가 만료된 orphan 케이스는 findById 측에서 빈 Map 으로 자연 필터링된다.

*/ @Override public List findAdmissionCandidates(ProgramId programId, int batchSize) { @@ -310,7 +284,7 @@ public List findAdmissionCandidates(ProgramId programId, int batchSi *
  • Hash 의 status / entryToken 업데이트
    • * * - *

    역인덱스는 유지 — 사용자가 GET 으로 자기 토큰 조회 가능 (status: ADMITTED 응답). + *

    역인덱스는 유지 — 사용자가 GET 으로 자기 토큰 조회 가능 (status: ADMITTED 응답).

    */ public void admit(QueueToken token) { String programKey = programKey(token.getProgramId()); @@ -327,148 +301,48 @@ public void admit(QueueToken token) { @Override public List execute(RedisOperations operations) { operations.multi(); - // 1. Sorted Set 에서 멤버 제거 (큐에서 빠짐) operations.opsForZSet().remove(programKey, tokenIdStr); - // 2. Hash 의 status / entryToken 업데이트 operations.opsForHash().putAll(tokenKey, updates); - return operations.exec(); } }); } -// program-service 와 kafka 이벤트 통합 후 교체 완료 -// -// /** -// * Redis 기반 findActiveProgramIds 구현. -// * -// *

    {@code queue:program:*} 패턴으로 SCAN 하여 활성 프로그램 ID 목록을 반환한다. -// * -// *

    SCAN 사용 이유: KEYS 명령은 production 에서 블로킹 발생 위험. SCAN 은 점진적. -// * -// *

    Future Work

    -// * 본 메서드는 MVP 의 가정 (큐 존재 = 활성 프로그램) 을 따른다. -// *

    program-service 와 Kafka 이벤트 통합 후엔: -// *

      -// *
    • {@code program.opened} 이벤트 → Redis Set 의 활성 프로그램 추가
      • -// *
    • {@code program.closed} 이벤트 → Set 에서 제거
      • -// *
    • 본 메서드는 Redis Set 직접 조회 (SCAN 불필요)
      • -// *
    -// */ -// public List findActiveProgramIds() { -// String pattern = QUEUE_KEY_PREFIX + PROGRAM_KEY_PREFIX + "*"; -// -// // 1. SCAN 으로 모든 큐 키 수집 -// Set keys = redisTemplate.execute((RedisCallback>) connection -> { -// Set result = new HashSet<>(); -// ScanOptions options = ScanOptions.scanOptions() -// .match(pattern) // queue:program:* 매칭 -// .count(100) // 한 번에 100 개씩 점진적 조회 -// .build(); -// -// // try-with-resources 로 cursor 자동 정리 -// try (Cursor cursor = connection.scan(options)) { -// while (cursor.hasNext()) { -// // Redis 는 byte[] 반환 → UTF-8 문자열로 변환 -// result.add(new String(cursor.next(), StandardCharsets.UTF_8)); -// } -// } -// return result; -// }); -// -// if (keys == null || keys.isEmpty()) { -// return List.of(); -// } -// -// // 2. 키에서 UUID 추출하여 ProgramId 로 변환 -// // 예: "queue:program:abc-123" → "abc-123" → ProgramId.of("abc-123") -// String prefix = QUEUE_KEY_PREFIX + PROGRAM_KEY_PREFIX; -// return keys.stream() -// .map(key -> key.substring(prefix.length())) -// .map(ProgramId::fromString) -// .toList(); -// } - /** - * Redis 기반 deleteAllByProgramId 구현. + * Redis 기반 deleteAllByProgramId 구현 (Lua 스크립트로 원자 처리). * - *

    Program 취소 시 해당 프로그램의 모든 토큰 (대기 + 입장) 정리. - * - *

    처리 단계: + *

    Lua 스크립트 안에서: *

      - *
    1. SCAN 으로 모든 token Hash 키 ({@code queue:token:*}) 조회
      2. - *
    2. 각 Hash 의 programId 필드를 확인하여 일치하는 토큰 식별
      3. - *
    3. 일치 토큰의 Hash + 역인덱스 + Sorted Set 자체를 일괄 삭제
      4. + *
    4. 프로그램 단위 토큰 인덱스 (Set) 의 멤버 조회 (SMEMBERS)
      5. + *
    5. 각 토큰별 compare-and-delete (역인덱스 + Hash)
      6. + *
    6. Sorted Set 자체 + seqKey + 프로그램 단위 인덱스 일괄 삭제
      7. *
    * - *

    주의: 입장 (ADMITTED) 된 토큰은 Sorted Set 에서 이미 빠져 있어 - * Sorted Set 조회로는 못 찾는다. ADMITTED 토큰 정리는 SCAN 으로 추가 처리. - * - *

    미래 개선: 프로그램별 토큰 ID Set 을 별도 유지하면 SCAN 불필요. - * 예: {@code queue:program:{programId}:all = {tokenId1, tokenId2, ...}}

    + *

    역인덱스 삭제는 compare-and-delete 로 TOCTOU 방어: + * 동시에 새 토큰이 같은 사용자로 enqueue 되어도 새 토큰의 역인덱스는 보존. */ @Override public void deleteAllByProgramId(ProgramId programId) { String programKey = programKey(programId); String programIdStr = programId.asString(); + String seqKey = seqKey(programId); + String programTokensKey = programTokensKey(programId); + + Long processedCount = redisTemplate.execute( + deleteAllByProgramScript, + List.of(programKey, seqKey, programTokensKey), + programIdStr, + QUEUE_KEY_PREFIX + TOKEN_KEY_PREFIX, // tokenKeyPrefix + QUEUE_KEY_PREFIX + USER_KEY_PREFIX, // userProgramKeyPrefix + ":" + PROGRAM_KEY_PREFIX // programKeyInfix + ); - // 1. SCAN 으로 모든 token Hash 키 찾기 (programId 일치하는 것) - Set tokenKeysToDelete = new HashSet<>(); - List userProgramKeysToDelete = new ArrayList<>(); - - String tokenKeyPattern = QUEUE_KEY_PREFIX + TOKEN_KEY_PREFIX + "*"; - Set allTokenKeys = scanKeys(tokenKeyPattern); - - HashOperations hashOps = redisTemplate.opsForHash(); - - for (String tokenKey : allTokenKeys) { - String tokenProgramId = hashOps.get(tokenKey, FIELD_PROGRAM_ID); - if (programIdStr.equals(tokenProgramId)) { - // 일치 → 삭제 대상 - tokenKeysToDelete.add(tokenKey); - - // 역인덱스 키도 만들기 위해 userId 조회 - String userId = hashOps.get(tokenKey, FIELD_USER_ID); - if (userId != null) { - String userProgramKey = QUEUE_KEY_PREFIX + USER_KEY_PREFIX + userId - + ":" + PROGRAM_KEY_PREFIX + programIdStr; - userProgramKeysToDelete.add(userProgramKey); - } - } - } - - // 2. 삭제 대상 모두 모음 (Sorted Set + Hash + 역인덱스) - List allKeysToDelete = new ArrayList<>(); - allKeysToDelete.add(programKey); // Sorted Set - allKeysToDelete.addAll(tokenKeysToDelete); - allKeysToDelete.addAll(userProgramKeysToDelete); - - // 3. 일괄 삭제 - Long deletedCount = redisTemplate.delete(allKeysToDelete); - - log.info("프로그램 토큰 삭제 완료. programId={}, 삭제 키 수={}", - programIdStr, deletedCount); - } - - private Set scanKeys(String pattern) { - return redisTemplate.execute((RedisCallback>) connection -> { - Set result = new HashSet<>(); - ScanOptions options = ScanOptions.scanOptions() - .match(pattern) - .count(100) - .build(); - try (Cursor cursor = connection.scan(options)) { - while (cursor.hasNext()) { - result.add(new String(cursor.next(), StandardCharsets.UTF_8)); - } - } - return result; - }); + log.info("프로그램 토큰 삭제 완료. programId={}, 처리 토큰 수={}", + programIdStr, processedCount); } - // ===== 키 생성 헬퍼 ===== private String programKey(ProgramId programId) { @@ -483,4 +357,12 @@ private String userProgramKey(UserId userId, ProgramId programId) { return QUEUE_KEY_PREFIX + USER_KEY_PREFIX + userId.asString() + ":" + PROGRAM_KEY_PREFIX + programId.asString(); } + + private String seqKey(ProgramId programId) { + return SEQ_KEY_PREFIX + programId.asString(); + } + + private String programTokensKey(ProgramId programId) { + return QUEUE_KEY_PREFIX + PROGRAM_KEY_PREFIX + programId.asString() + PROGRAM_TOKENS_SUFFIX; + } } diff --git a/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisScriptConfig.java b/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisScriptConfig.java new file mode 100644 index 0000000..a52b401 --- /dev/null +++ b/src/main/java/com/firstticket/queueservice/queuetoken/infrastructure/redis/RedisScriptConfig.java @@ -0,0 +1,52 @@ +package com.firstticket.queueservice.queuetoken.infrastructure.redis; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.ClassPathResource; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.scripting.support.ResourceScriptSource; + +/** + * Redis Lua 스크립트 빈 등록. + * + *

    {@code DefaultRedisScript} 는 스크립트 본문을 캐시하고 SHA-1 으로 Redis 에 EVALSHA 호출하여 + * 매 호출마다 본문 전송을 피한다.

    + */ +@Configuration +public class RedisScriptConfig { + /** + * enqueue 원자 처리 스크립트. + * 반환: Long (1=성공, 0=중복) + */ + @Bean + public DefaultRedisScript enqueueScript() { + DefaultRedisScript script = new DefaultRedisScript<>(); + script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/enqueue.lua"))); + script.setResultType(Long.class); + return script; + } + + /** + * delete 원자 처리 스크립트. + * 반환: Long (1=역인덱스 같이 삭제, 2=역인덱스 보존) + */ + @Bean + public DefaultRedisScript deleteScript() { + DefaultRedisScript script = new DefaultRedisScript<>(); + script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/delete.lua"))); + script.setResultType(Long.class); + return script; + } + + /** + * deleteAllByProgram 원자 처리 스크립트 + * 반환: Long + */ + @Bean + public DefaultRedisScript deleteAllByProgramScript() { + DefaultRedisScript script = new DefaultRedisScript<>(); + script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/deleteAllByProgram.lua"))); + script.setResultType(Long.class); + return script; + } +} diff --git a/src/main/resources/lua/delete.lua b/src/main/resources/lua/delete.lua new file mode 100644 index 0000000..d4b9257 --- /dev/null +++ b/src/main/resources/lua/delete.lua @@ -0,0 +1,47 @@ +-- delete.lua +-- 대기 토큰 삭제 (원자적) +-- +-- 흐름: +-- 1. 역인덱스 compare-and-delete +-- 2. Sorted Set 멤버 제거 (ZREM) +-- 3. Hash 키 삭제 (DEL) +-- 4. 프로그램 단위 토큰 인덱스에서 제거 (SREM) +-- +-- KEYS: +-- [1] userProgramKey +-- [2] programKey +-- [3] tokenKey +-- [4] programTokensKey ← 새 본문 +-- +-- ARGV: +-- [1] tokenId + +local userProgramKey = KEYS[1] +local programKey = KEYS[2] +local tokenKey = KEYS[3] +local programTokensKey = KEYS[4] + +local tokenId = ARGV[1] + +-- 1. 역인덱스 compare-and-delete +local current = redis.call('GET', userProgramKey) +local indexDeleted = 0 +if current == tokenId then + redis.call('DEL', userProgramKey) + indexDeleted = 1 +end + +-- 2. Sorted Set 멤버 제거 +redis.call('ZREM', programKey, tokenId) + +-- 3. Hash 삭제 +redis.call('DEL', tokenKey) + +-- 4. 프로그램 단위 토큰 인덱스에서 제거 +redis.call('SREM', programTokensKey, tokenId) + +if indexDeleted == 1 then + return 1 +else + return 2 +end diff --git a/src/main/resources/lua/deleteAllByProgram.lua b/src/main/resources/lua/deleteAllByProgram.lua new file mode 100644 index 0000000..6fe528d --- /dev/null +++ b/src/main/resources/lua/deleteAllByProgram.lua @@ -0,0 +1,64 @@ +-- deleteAllByProgram.lua +-- 프로그램 취소 시 해당 프로그램의 모든 토큰 일괄 삭제 (원자적) +-- +-- 흐름: +-- 1. 프로그램 단위 토큰 인덱스 (Set) 의 멤버 조회 (SMEMBERS) +-- 2. 각 토큰별 compare-and-delete: +-- - 역인덱스 (현재 값 일치 시만 DEL) +-- - Hash DEL +-- 3. Sorted Set 자체 + seqKey + 프로그램 단위 인덱스 일괄 DEL +-- +-- SCAN 사용 X — 프로그램 단위 인덱스로 토큰 범위 제한 (성능 + 블로킹 회피) +-- +-- KEYS: +-- [1] programKey (queue:program:{programId}) +-- [2] seqKey (queue:seq:{programId}) +-- [3] programTokensKey (queue:program:{programId}:tokens) +-- +-- ARGV: +-- [1] programIdStr +-- [2] tokenKeyPrefix (queue:token:) +-- [3] userProgramKeyPrefix (queue:user:) +-- [4] programKeyInfix (:program:) +-- +-- 반환: 처리된 토큰 수 + +local programKey = KEYS[1] +local seqKey = KEYS[2] +local programTokensKey = KEYS[3] + +local programIdStr = ARGV[1] +local tokenKeyPrefix = ARGV[2] +local userProgramKeyPrefix = ARGV[3] +local programKeyInfix = ARGV[4] + +-- 1. 프로그램 단위 토큰 인덱스 조회 (SCAN X) +local tokenIds = redis.call('SMEMBERS', programTokensKey) +local processedCount = 0 + +-- 2. 각 토큰별 정리 +for i, tokenIdStr in ipairs(tokenIds) do + local tokenKey = tokenKeyPrefix .. tokenIdStr + local userId = redis.call('HGET', tokenKey, 'userId') + + -- 역인덱스 compare-and-delete + if userId then + local userProgramKey = userProgramKeyPrefix .. userId .. programKeyInfix .. programIdStr + local currentTokenIdInIndex = redis.call('GET', userProgramKey) + if currentTokenIdInIndex == tokenIdStr then + redis.call('DEL', userProgramKey) + end + end + + -- Hash 삭제 + redis.call('DEL', tokenKey) + + processedCount = processedCount + 1 +end + +-- 3. Sorted Set 자체 + seqKey + 프로그램 단위 인덱스 일괄 삭제 +redis.call('DEL', programKey) +redis.call('DEL', seqKey) +redis.call('DEL', programTokensKey) + +return processedCount diff --git a/src/main/resources/lua/enqueue.lua b/src/main/resources/lua/enqueue.lua new file mode 100644 index 0000000..777ca34 --- /dev/null +++ b/src/main/resources/lua/enqueue.lua @@ -0,0 +1,70 @@ +-- enqueue.lua +-- 대기 토큰 진입 (원자적) +-- +-- 흐름: +-- 1. 역인덱스 (SETNX) 로 중복 진입 방지 +-- 2. Score 생성 (epoch_second * 1000000 + INCR) — tie-breaker +-- 3. Sorted Set 추가 (ZADD) +-- 4. Hash 메타 저장 (HSET) + TTL (EXPIRE) +-- 5. 프로그램 단위 토큰 인덱스 (Set) 추가 (SADD) +-- +-- KEYS: +-- [1] userProgramKey +-- [2] programKey +-- [3] tokenKey +-- [4] seqKey +-- [5] programTokensKey +-- +-- ARGV: +-- [1] tokenId +-- [2] userId +-- [3] programId +-- [4] issuedAtEpochMilli +-- [5] status +-- [6] ttlSeconds +-- +-- 반환: +-- 1 = 성공 +-- 0 = 이미 존재 (중복 진입) + +local userProgramKey = KEYS[1] +local programKey = KEYS[2] +local tokenKey = KEYS[3] +local seqKey = KEYS[4] +local programTokensKey = KEYS[5] + +local tokenId = ARGV[1] +local userId = ARGV[2] +local programId = ARGV[3] +local issuedAtEpochMilli = tonumber(ARGV[4]) +local status = ARGV[5] +local ttlSeconds = tonumber(ARGV[6]) + +-- 1. 역인덱스 (SETNX) — 같은 user+program 토큰 이미 있으면 실패 +local acquired = redis.call('SET', userProgramKey, tokenId, 'NX', 'EX', ttlSeconds) +if not acquired then + return 0 +end + +-- 2. tie-breaker score 생성 (epoch_second * 1000000 + 시퀀스) +local seq = redis.call('INCR', seqKey) +redis.call('EXPIRE', seqKey, ttlSeconds) +local issuedAtEpochSecond = math.floor(issuedAtEpochMilli / 1000) +local score = issuedAtEpochSecond * 1000000 + seq + +-- 3. Sorted Set 추가 +redis.call('ZADD', programKey, score, tokenId) + +-- 4. Hash 메타 저장 + TTL +redis.call('HSET', tokenKey, + 'userId', userId, + 'programId', programId, + 'issuedAt', tostring(issuedAtEpochMilli), + 'status', status) +redis.call('EXPIRE', tokenKey, ttlSeconds) + +-- 5. 프로그램 단위 토큰 인덱스 (Set) 추가 +redis.call('SADD', programTokensKey, tokenId) +redis.call('EXPIRE', programTokensKey, ttlSeconds) + +return 1