diff --git a/ai/journal/2026/08/AIJ-0021-sweep.md b/ai/journal/2026/08/AIJ-0021-sweep.md new file mode 100644 index 00000000..2a97dbf6 --- /dev/null +++ b/ai/journal/2026/08/AIJ-0021-sweep.md @@ -0,0 +1,136 @@ +--- +id: AIJ-0021 +date: 2026-08-20 +kind: implement +phase: 3 +plan: [3.6.1, 3.6.2, 3.6.3, 3.6.4] +jira: CY-233 +commits: [] +agent: claude-opus-5 +confidence: medium +promoted-to: +--- + +# 이탈자 청소 — 앞부분만 본다 + +## 무엇을 + +`sweep.lua`. 생존 신호가 끊긴 앞부분 K 명을 큐에서 빼고 유예 기록으로 옮긴다. +만료된 기록은 함께 지운다. + +## 왜 (근거) + +**전체를 훑지 않는다.** 2만 명 큐에서 매 틱 전원을 보면 **청소 자체가 부하**다. +앞부분만 봐도 되는 이유는 이탈이 앞에서부터 드러나기 때문이다 — 뒤엣사람은 +아직 폴링할 차례가 안 왔을 뿐 죽은 것이 아니다. + +**제거와 기록이 갈리면** 자리도 잃고 재방문자로도 식별 안 되는 사람이 생긴다. +한쪽만 성공한 상태를 없애려고 같은 스크립트에 둔다. + +**자리는 보관하지 않는다** (D-11). 유예 기록은 재방문자 식별용이고, 돌아오면 +줄 맨 뒤에 선다. 보관하면 이탈자가 돌아올 때마다 성실히 기다린 사람이 밀린다. + +**유예 해시가 무한히 자라지 않게 한다** (RD-7). 해시는 **필드별 TTL 을 못 +걸어서** 값에 시각을 담고 이 스크립트가 직접 지운다. 청소하는 김에 같이 한다 — +따로 도는 것을 하나 더 만들면 그것도 관리 대상이다. + +**값이 깨진 기록도 지운다.** 숫자가 아니면 언제 것인지 알 수 없고, 남겨 두면 +**영원히 안 지워진다.** + +**시각을 주입받는다.** `TIME` 을 쓰면 복제본마다 다른 값을 보고, 테스트에서 +만료를 확인할 수도 없다 (TS-4). + +## 고려했으나 택하지 않은 것 + +- **전체 훑기** — 정확하다. 그런데 2만 명 × 쿠폰 수를 매 틱 보는 것은 + 대기열이 커질수록 나빠진다. K 를 인자로 받아 **부하와 정확도를 운영자가 + 맞바꾸게** 했다 (P-1). +- **유예 만료를 별도 작업으로** — 청소와 무관해 보인다. 그런데 그러면 도는 + 것이 하나 더 늘고, 둘 다 같은 해시를 만져 경합이 생긴다. +- **`alive` 없는 사람을 즉시 삭제** — 유예를 안 남기면 재방문자를 식별할 수 + 없다. 그건 Phase 7 의 재방문 처리를 통째로 막는다. + +## 리뷰가 짚은 것 — 상한이 절반만 걸려 있었다 + +**`K` 를 작게 줘도 어딘가에서 전체를 훑으면 그 `K` 는 아무 의미가 없다.** +처음 판은 `ZRANGEBYSCORE` 로 살아 있는 쪽을 **전부** 받아 집합을 만들었다. +`K=1` 이어도 `alive` ZSET 크기에 비례해 이벤트 루프를 잡는다. + +`ZMSCORE` 로 **앞부분 K 명의 score 만** 묻는다. 만료 신호 정리도 +`ZRANGE ... BYSCORE LIMIT` 으로 예산 안에서만 걷는다. + +**`HSCAN COUNT` 는 상한이 아니라 힌트다.** 해시가 조밀하게 인코딩돼 있으면 +한 번에 `budget` 보다 많이 돌아온다. 받은 것을 다 지우면 예산이 있으나 마나라, +**받은 것 중 예산만큼만** 지우고 나머지는 다음 호출로 미룬다. + +**커서도 쓰기 전에 검증한다.** 형식이 틀리면 `HSCAN` 이 오류를 내는데 그때는 +이미 앞의 쓰기가 끝나 있다. Lua 는 롤백하지 않는다 — 이 페이즈에서 같은 +실수를 세 번째 한 자리다. + +## 검사가 검사 구실을 못 하던 것 + +**`KEYS` 정적 검사가 지역 변수를 무조건 허용했다.** +`local rogue = 'queue:{c1}'` 를 거쳐 들어오면 통과한다 — 검사가 있는데 +아무것도 안 보는 상태다. `KEYS[n]` 이 **대입된** 변수만 추적하게 고쳤다. + +**계약 주석 검사가 파일 전체에서 `KEYS[1]` 을 찾았다.** 코드에서 쓰는 것과 +계약을 적어 둔 것은 다르다. 상단 주석 블록만 보고 `ARGV[1]` 도 함께 본다. + +**하드코딩 키 검사가 줄 주석만 걷어냈다.** Javadoc 안의 예시가 위반으로 +잡힌다 — 오탐이 나면 사람은 검사를 고치는 대신 우회한다. + +## 확신이 낮은 부분 + +- **K 의 기본값이 없다.** 인자로 받게만 해 뒀다. 무엇에서 나와야 하는지는 + 아래에 적어 뒀지만 대입할 수치가 실측 전이다. +- **커서를 호출부가 들고 있어야 한다.** 스크립트가 다음 커서를 돌려주지만 + 그걸 다음 틱에 넘기는 것은 스케줄러 몫이다 (Phase 4). 안 넘기면 앞부분만 + 계속 보고 뒤는 영영 안 지워진다. + +## K 를 무엇으로 정하는가 + +숫자는 없지만 기준은 셋이다. **하한 하나와 상한 둘**이고, 상한 중 하나는 +성능이 아니라 정합성 제약이라 이쪽이 먼저 걸린다. + +**하한 — 다음 틱에 입장시킬 구간은 덮어야 한다.** +이탈자가 큐에 남아도 뒷사람을 막지는 않는다. 입장은 pop 이 아니라 +`admitted` 워터마크 비교라 그냥 지나간다. 대신 **워터마크가 유령 위를 +지나가면 그 입장분이 버려진다** — 뒷단 가용량을 안 온 사람에게 쓴 것이다. +그러니 봐야 할 구간은 "앞에서 적당히" 가 아니라 **배분기가 다음 틱에 +워터마크를 밀어 넘길 그 구간**이다. + + K ≳ (틱당 배분량) × (1 + 여유) + +여유는 이탈자가 뭉친 경우와 틱 지터를 흡수한다. 그리고 **청소가 배분보다 +먼저** 같은 틱 안에서 돌아야 한다 — 순서가 뒤집히면 이번 틱은 유령에게 +배분하고 다음 틱에 치운다. 이보다 깊이 봐도 이득은 적다. 더 뒤의 이탈자는 +앞으로 올라오며 걸리고, 그전까지는 ETA 만 부풀릴 뿐 용량을 안 먹는다. + +**상한 1 — 밴드 경계를 넘으면 단일 임계가 성립하지 않는다.** +`alive` score 는 사람마다 자기 밴드의 TTL 로 찍히는데 (`PollIntervalPolicy` +기준 1·3·10·30 초, TTL 은 그 3 배에 하한 30 초), 청소는 `at < now` 하나로만 +판정한다. 그 판정이 옳으려면 **보는 구간의 폴링 리듬이 균질해야 한다.** +K 가 ETA 5 초 밴드를 넘어 내려가면 30 초마다 두드리는 게 정상인 사람이 같은 +잣대에 걸린다 — 불변식 4 가 깨지는 자리다. **K 는 가장 앞 밴드 안에 머무는 +것이 안전선**이고, 넘겨야 한다면 K 를 키울 게 아니라 밴드별 임계를 나눈다. + +**상한 2 — 스크립트 하나가 이벤트 루프를 잡는 시간.** +`ZRANGE K` + `ZMSCORE K` + `ZREM` 이라 O(K) 다. 실무적으로 가장 헐거운 +제약이다. 20,000 동시 대기(R4)에서 나올 K 는 수백~수천이고 그 정도는 +서브밀리초라, 상한 1 에 먼저 걸린다. + +**재려면 둘이 필요하다** — 이탈률, 그리고 실제 폴링 간격 분포. 후자는 +백그라운드 탭이 분당 1 회로 스로틀되는 것이 `MIN_ALIVE_TTL_SEC = 30` 의 +근거인데 그 30 이 충분한지가 아직 미검증이다. + +## 검증 + +- 통합 9건 전건 통과 +- 앞부분만 제거 · 생존 신호 있으면 유지 · K 가 인자로 동작 +- 제거와 기록이 함께 · 만료·깨진 값 정리 · 잘못된 인자에 무변경 + +## 다음 사람에게 + +**"전체를 보면 정확하다" 가 늘 맞지는 않는다.** 이 청소는 매 틱 도는 배경 +작업이라, 정확도를 조금 얻자고 부하를 크게 치르면 그 부하가 다시 이탈을 +만든다. 앞부분만 보는 것은 타협이 아니라 **부하를 유계로 두려는 설계**다. diff --git a/ai/journal/index.md b/ai/journal/index.md index 8975eaec..47eb7bfb 100644 --- a/ai/journal/index.md +++ b/ai/journal/index.md @@ -8,6 +8,7 @@ | ID | 날짜 | 종류 | 제목 | 확신 | 승격 | |---|---|---|---|---|---| +| [AIJ-0021](2026/08/AIJ-0021-sweep.md) | 2026-08-20 | implement | 이탈자 청소 — 앞부분만 본다 | medium | — | | [AIJ-0020](2026/08/AIJ-0020-queue-status.md) | 2026-08-19 | implement | 순번 조회 — 한 번에 끝나야 하는 이유 | high | — | | [AIJ-0019](2026/08/AIJ-0019-library-first.md) | 2026-08-19 | decision | 라이브러리를 먼저 본다 — 안 쓴 이유를 남긴다 | medium | DS-8 | | [AIJ-0018](2026/08/AIJ-0018-enqueue.md) | 2026-08-19 | implement | 큐 등록 — 순서가 곧 정책이다 | high | — | diff --git a/src/main/java/com/kafkick/waiting/adapter/redis/RedisKeys.java b/src/main/java/com/kafkick/waiting/adapter/redis/RedisKeys.java index 36f63c97..5ef7cebb 100644 --- a/src/main/java/com/kafkick/waiting/adapter/redis/RedisKeys.java +++ b/src/main/java/com/kafkick/waiting/adapter/redis/RedisKeys.java @@ -52,9 +52,16 @@ public static String grace(String couponId, int shards, int shard) { return "grace:{" + tag(couponId, shards, shard) + "}"; } - /** 생존 신호. 폴링이 곧 하트비트다. */ - public static String alive(String couponId, int shards, int shard, String memberId) { - return "alive:{" + tag(couponId, shards, shard) + "}:" + validated(memberId, "memberId"); + /** + * 생존 신호. 폴링이 곧 하트비트다. + * + *

사람마다 키를 만들지 않는다. 그러면 청소 스크립트가 KEYS 에 + * 선언되지 않은 키를 만지게 되고 클러스터가 거부한다 (RD-1). 쿠폰당 ZSET + * 하나에 만료 시각을 score 로 담는다 — 개별 TTL 은 잃지만 청소가 + * 어차피 만료를 보므로 잃는 것이 없다. + */ + public static String alive(String couponId, int shards, int shard) { + return "alive:{" + tag(couponId, shards, shard) + "}"; } /** diff --git a/src/main/resources/redis/enqueue.lua b/src/main/resources/redis/enqueue.lua index 37e4793e..c440d818 100644 --- a/src/main/resources/redis/enqueue.lua +++ b/src/main/resources/redis/enqueue.lua @@ -2,11 +2,12 @@ -- -- KEYS[1] queue:{cid} ZSET. score = Redis TIME 의 마이크로초 -- KEYS[2] maxscore:{cid} 시계 역행 방어용 바닥값 --- KEYS[3] alive:{cid}:{member} 생존 신호. 폴링이 곧 하트비트다 +-- KEYS[3] alive:{cid} 생존 신호 ZSET. score 는 만료 시각(초) -- ARGV[1] memberId -- ARGV[2] maxscore TTL(초). 양의 정수 -- ARGV[3] alive TTL(초). 양의 정수. 폴링 간격에서 나온 값이라 주입받는다 -- ARGV[4] 큐 길이 상한. 0 이면 상한 없음 +-- ARGV[5] 지금 시각(초). 생존 신호의 만료 시각을 계산한다 -- -- 반환 {score, floorApplied, alreadyQueued} -- score 이 사람의 순번. 거부되면 '-1' @@ -37,6 +38,11 @@ local aliveTtl aliveTtl, err = positive_int(ARGV[3], 'alive TTL') if not aliveTtl then return redis.error_reply(err) end +local now = tonumber(ARGV[5]) +if now == nil or now < 0 then + return redis.error_reply('시각은 0 이상이어야 한다: ' .. tostring(ARGV[5])) +end + local maxLen = tonumber(ARGV[4]) if maxLen == nil or maxLen < 0 or maxLen ~= math.floor(maxLen) then return redis.error_reply('큐 길이 상한은 0 이상 정수여야 한다: ' .. tostring(ARGV[4])) @@ -49,7 +55,7 @@ end -- 것이 그 사람 잘못이 아닌데 그가 자리를 잃는다. local existing = redis.call('ZSCORE', KEYS[1], ARGV[1]) if existing then - redis.call('SET', KEYS[3], '1', 'EX', aliveTtl) + redis.call('ZADD', KEYS[3], now + aliveTtl, ARGV[1]) return {existing, 0, 1} end @@ -58,8 +64,10 @@ if maxLen > 0 and redis.call('ZCARD', KEYS[1]) >= maxLen then return {'-1', 0, 0} end -local now = redis.call('TIME') -local score = tonumber(now[1]) * 1000000 + tonumber(now[2]) +-- 이름을 now 와 겹치지 않게 둔다. 주입받은 시각(초)과 Redis 시계(μs)는 +-- 단위도 출처도 다른 값이라 한 이름을 쓰면 조용히 섞인다. +local redisTime = redis.call('TIME') +local score = tonumber(redisTime[1]) * 1000000 + tonumber(redisTime[2]) local floor = tonumber(redis.call('GET', KEYS[2]) or 0) local applied = 0 @@ -78,6 +86,6 @@ end -- 그건 maxmemory 로 막는다. redis.call('ZADD', KEYS[1], score, ARGV[1]) redis.call('SET', KEYS[2], score, 'EX', scoreTtl) -redis.call('SET', KEYS[3], '1', 'EX', aliveTtl) +redis.call('ZADD', KEYS[3], now + aliveTtl, ARGV[1]) return {tostring(score), applied, 0} diff --git a/src/main/resources/redis/queue_status.lua b/src/main/resources/redis/queue_status.lua index ff27929a..d4aa35e8 100644 --- a/src/main/resources/redis/queue_status.lua +++ b/src/main/resources/redis/queue_status.lua @@ -2,10 +2,11 @@ -- -- KEYS[1] queue:{cid} ZSET -- KEYS[2] admitted:{cid} 입장 임계. 개수가 아니라 score 값이다 (D-8) --- KEYS[3] alive:{cid}:{member} 생존 신호 +-- KEYS[3] alive:{cid} 생존 신호 ZSET. score 는 만료 시각(초) -- KEYS[4] grace:{cid} 이탈 기록 해시 -- ARGV[1] memberId -- ARGV[2] alive TTL(초). 양의 정수 +-- ARGV[3] 지금 시각(초) -- -- 반환 {state, rank, score} -- state 'WAITING' | 'ADMITTED' | 'NOT_QUEUED' @@ -21,6 +22,11 @@ if ttl == nil or ttl < 1 or ttl ~= math.floor(ttl) then return redis.error_reply('alive TTL 은 양의 정수여야 한다: ' .. tostring(ARGV[2])) end +local now = tonumber(ARGV[3]) +if now == nil or now < 0 then + return redis.error_reply('시각은 0 이상이어야 한다: ' .. tostring(ARGV[3])) +end + local score = redis.call('ZSCORE', KEYS[1], ARGV[1]) if not score then -- **0번째와 구분한다.** 없는 것과 맨 앞인 것은 다르다. 뭉치면 유실된 @@ -29,12 +35,13 @@ if not score then end -- 폴링이 곧 생존 신호다. 조회한 김에 갱신한다 — 왕복을 늘리지 않는다. -redis.call('SET', KEYS[3], '1', 'EX', ttl) +redis.call('ZADD', KEYS[3], now + ttl, ARGV[1]) local admitted = tonumber(redis.call('GET', KEYS[2]) or -1) if admitted >= 0 and tonumber(score) <= admitted then -- 차례가 왔다. 큐에서 빼지 않으면 대기 인원이 계속 부풀고 ETA 가 틀어진다. redis.call('ZREM', KEYS[1], ARGV[1]) + redis.call('ZREM', KEYS[3], ARGV[1]) redis.call('HSET', KEYS[4], ARGV[1], 'admitted') return {'ADMITTED', 0, score} end diff --git a/src/main/resources/redis/sweep.lua b/src/main/resources/redis/sweep.lua new file mode 100644 index 00000000..49a9b958 --- /dev/null +++ b/src/main/resources/redis/sweep.lua @@ -0,0 +1,111 @@ +-- 이탈자 청소. **한 번의 실행이 하는 일에 상한을 둔다.** +-- +-- KEYS[1] queue:{cid} ZSET +-- KEYS[2] grace:{cid} 이탈 기록 해시 +-- KEYS[3] alive:{cid} 생존 신호 ZSET. score 는 만료 시각(초) +-- ARGV[1] 검사 범위 K. 큐 앞에서 이만큼만 본다 +-- ARGV[2] 지금 시각(초). 도메인처럼 주입받는다 +-- ARGV[3] 유예 보관 기간(초) +-- ARGV[4] 정리 예산. 만료 신호와 유예 기록을 각각 이만큼까지 지운다 +-- ARGV[5] HSCAN 커서. 첫 호출은 '0'. 반환된 값을 다음에 넘긴다 +-- +-- 반환 {swept, expiredSignals, expiredGrace, nextCursor} +-- +-- **키를 문자열로 조립하지 않는다.** 사람마다 alive 키를 만들면 KEYS 에 +-- 선언되지 않은 키를 만지게 되고 클러스터가 거부한다 (RD-1). +-- +-- **모든 순회에 상한이 걸려 있어야 한다.** Lua 는 통째로 도는 동안 다른 +-- 요청을 전부 막는다. K 를 작게 줘도 어딘가에서 전체를 훑으면 그 K 는 +-- 아무 의미가 없다. + +local limit = tonumber(ARGV[1]) +if limit == nil or limit < 1 or limit ~= math.floor(limit) then + return redis.error_reply('검사 범위는 양의 정수여야 한다: ' .. tostring(ARGV[1])) +end + +local now = tonumber(ARGV[2]) +if now == nil or now < 0 then + return redis.error_reply('시각은 0 이상이어야 한다: ' .. tostring(ARGV[2])) +end + +local retention = tonumber(ARGV[3]) +if retention == nil or retention < 1 or retention ~= math.floor(retention) then + return redis.error_reply('유예 보관 기간은 양의 정수여야 한다: ' .. tostring(ARGV[3])) +end + +local budget = tonumber(ARGV[4]) +if budget == nil or budget < 1 or budget ~= math.floor(budget) then + return redis.error_reply('정리 예산은 양의 정수여야 한다: ' .. tostring(ARGV[4])) +end + +-- **커서도 쓰기 전에 본다.** 형식이 틀리면 HSCAN 이 오류를 내는데, 그때는 +-- 이미 앞의 쓰기가 끝나 있다. Lua 는 롤백하지 않는다. +local cursor = ARGV[5] +if cursor == nil or not string.match(cursor, '^%d+$') then + return redis.error_reply('커서는 숫자여야 한다: ' .. tostring(cursor)) +end + +-- 앞에서 K 명. 뒤는 아직 볼 때가 아니다. +local front = redis.call('ZRANGE', KEYS[1], 0, limit - 1) +local swept = 0 + +if #front > 0 then + -- **앞부분의 score 만 묻는다.** ZRANGEBYSCORE 로 살아 있는 쪽을 다 받으면 + -- K 를 1 로 줘도 alive 전체 크기에 비례해 이벤트 루프를 잡는다. + local scores = redis.call('ZMSCORE', KEYS[3], unpack(front)) + + local gone = {} + local records = {} + for i = 1, #front do + -- score 가 없거나 이미 지난 것은 폴링이 끊긴 것이다 + local at = tonumber(scores[i]) + if at == nil or at < now then + gone[#gone + 1] = front[i] + -- 자리는 안 보관한다. 재방문자로 식별만 한다 (D-11). + records[#records + 1] = front[i] + records[#records + 1] = now + end + end + + if #gone > 0 then + redis.call('ZREM', KEYS[1], unpack(gone)) + redis.call('HSET', KEYS[2], unpack(records)) + swept = #gone + end +end + +-- 만료된 생존 신호도 예산 안에서만 걷는다. ZREMRANGEBYSCORE 는 대상 수만큼 +-- 도므로 한 번에 다 지우려 하면 그 자체가 오래 걸린다. +local staleSignals = redis.call('ZRANGE', KEYS[3], '-inf', '(' .. now, + 'BYSCORE', 'LIMIT', 0, budget) +local expiredSignals = 0 +if #staleSignals > 0 then + redis.call('ZREM', KEYS[3], unpack(staleSignals)) + expiredSignals = #staleSignals +end + +-- **COUNT 는 힌트지 상한이 아니다.** 해시가 조밀하게 인코딩돼 있으면 한 번에 +-- budget 보다 많이 돌아온다. 받은 것 중 예산만큼만 지우고 나머지는 다음 +-- 호출로 미룬다 — 커서만 넘기면 초과분을 막을 수 없다. +local scanned = redis.call('HSCAN', KEYS[2], cursor, 'COUNT', budget) +local fields = scanned[2] +local cutoff = now - retention +local doomed = {} +for i = 1, #fields, 2 do + if #doomed >= budget then + break + end + local at = tonumber(fields[i + 1]) + if at == nil or at < cutoff then + doomed[#doomed + 1] = fields[i] + end +end + +local expiredGrace = 0 +if #doomed > 0 then + redis.call('HDEL', KEYS[2], unpack(doomed)) + expiredGrace = #doomed +end + +-- 다음 커서를 돌려준다. 호출부가 이어서 넘긴다. +return {swept, expiredSignals, expiredGrace, scanned[1]} diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/ClockMonotonicTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/ClockMonotonicTest.java index 3e836177..c10816bc 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/ClockMonotonicTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/ClockMonotonicTest.java @@ -25,6 +25,8 @@ @SpringBootTest class ClockMonotonicTest extends RedisContainerSupport { + private static final String NOW = "1800000000"; + private static final long TTL_SECONDS = 86_400; private static final Duration WAIT = Duration.ofSeconds(5); @@ -54,12 +56,12 @@ private List enqueue(String memberId) { return (List) redis.execute( script, List.of(QUEUE, MAX_SCORE, alive(memberId)), - List.of(memberId, String.valueOf(TTL_SECONDS), ALIVE_TTL, NO_CAP)) + List.of(memberId, String.valueOf(TTL_SECONDS), ALIVE_TTL, NO_CAP, NOW)) .blockFirst(WAIT); } private String alive(String memberId) { - return RedisKeys.alive(COUPON, 1, 0, memberId); + return RedisKeys.alive(COUPON, 1, 0); } private long scoreOf(String memberId) { @@ -183,7 +185,7 @@ private long alreadyQueued(List result) { // maxscore 없는 ZSET 이 남아 "같이 남거나 같이 사라진다" 가 깨진다. org.assertj.core.api.Assertions.assertThatThrownBy(() -> redis.execute(script, List.of(QUEUE, MAX_SCORE, alive("m1")), - List.of("m1", "0", ALIVE_TTL, NO_CAP)) + List.of("m1", "0", ALIVE_TTL, NO_CAP, NOW)) .blockFirst(WAIT)) .rootCause() .hasMessageContaining("TTL"); diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/ConcurrentEnqueueTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/ConcurrentEnqueueTest.java index b306f1f2..fe15a8f1 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/ConcurrentEnqueueTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/ConcurrentEnqueueTest.java @@ -34,6 +34,8 @@ @SpringBootTest class ConcurrentEnqueueTest extends RedisContainerSupport { + private static final String NOW = "1800000000"; + private static final long TTL_SECONDS = 86_400; private static final Duration WAIT = Duration.ofSeconds(10); @@ -55,12 +57,12 @@ class ConcurrentEnqueueTest extends RedisContainerSupport { } private String alive(String memberId) { - return RedisKeys.alive(COUPON, 1, 0, memberId); + return RedisKeys.alive(COUPON, 1, 0); } private void enqueue(String memberId) { redis.execute(script, List.of(QUEUE, MAX_SCORE, alive(memberId)), - List.of(memberId, String.valueOf(TTL_SECONDS), ALIVE_TTL, NO_CAP)) + List.of(memberId, String.valueOf(TTL_SECONDS), ALIVE_TTL, NO_CAP, NOW)) .blockFirst(WAIT); } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/EnqueueGuardTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/EnqueueGuardTest.java index c30a931e..7789429e 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/EnqueueGuardTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/EnqueueGuardTest.java @@ -25,10 +25,13 @@ @SpringBootTest class EnqueueGuardTest extends RedisContainerSupport { + private static final String NOW = "1800000000"; + private static final Duration WAIT = Duration.ofSeconds(5); private static final String COUPON = "guard"; private static final String QUEUE = RedisKeys.queue(COUPON, 1, 0); private static final String MAX_SCORE = RedisKeys.maxScore(COUPON, 1, 0); + private static final String ALIVE = RedisKeys.alive(COUPON, 1, 0); @Autowired private ReactiveStringRedisTemplate redis; @@ -40,7 +43,7 @@ class EnqueueGuardTest extends RedisContainerSupport { script = RedisScript.of(new ClassPathResource("redis/enqueue.lua"), List.class); redis.delete(QUEUE, MAX_SCORE).block(WAIT); for (int i = 0; i < 10; i++) { - redis.delete(RedisKeys.alive(COUPON, 1, 0, "m" + i)).block(WAIT); + redis.delete(RedisKeys.alive(COUPON, 1, 0)).block(WAIT); } } @@ -48,18 +51,20 @@ class EnqueueGuardTest extends RedisContainerSupport { private List enqueue(String memberId, String aliveTtl, String cap) { return (List) redis.execute( script, - List.of(QUEUE, MAX_SCORE, RedisKeys.alive(COUPON, 1, 0, memberId)), - List.of(memberId, "86400", aliveTtl, cap)) + List.of(QUEUE, MAX_SCORE, RedisKeys.alive(COUPON, 1, 0)), + List.of(memberId, "86400", aliveTtl, cap, NOW)) .blockFirst(WAIT); } @Test - @DisplayName("등록하면_생존_키가_TTL과_함께_생긴다") - void 등록하면_생존_키가_TTL과_함께_생긴다() { + @DisplayName("등록하면_생존_신호에_만료_시각이_찍힌다") + void 등록하면_생존_신호에_만료_시각이_찍힌다() { + // ZSET 의 score 가 만료 시각이다. 사람마다 키를 만들면 청소가 + // KEYS 에 없는 키를 만지게 되고 클러스터가 거부한다 (RD-1). enqueue("m1", "30", "0"); - assertThat(redis.getExpire(RedisKeys.alive(COUPON, 1, 0, "m1")).block(WAIT)) - .isBetween(Duration.ofSeconds(25), Duration.ofSeconds(30)); + assertThat(redis.opsForZSet().score(ALIVE, "m1").block(WAIT)) + .isEqualTo(Long.parseLong(NOW) + 30); } @Test @@ -68,8 +73,8 @@ private List enqueue(String memberId, String aliveTtl, String cap) { // 폴링 간격에서 나오는 값이라 스크립트에 박으면 둘이 갈라진다. enqueue("m2", "90", "0"); - assertThat(redis.getExpire(RedisKeys.alive(COUPON, 1, 0, "m2")).block(WAIT)) - .isBetween(Duration.ofSeconds(85), Duration.ofSeconds(90)); + assertThat(redis.opsForZSet().score(ALIVE, "m2").block(WAIT)) + .isEqualTo(Long.parseLong(NOW) + 90); } @Test @@ -78,12 +83,12 @@ private List enqueue(String memberId, String aliveTtl, String cap) { // 순번은 그대로지만 살아 있다는 신호는 새로 찍혀야 한다. // 안 그러면 성실히 새로고침하는 사람이 이탈자로 지워진다. enqueue("m3", "30", "0"); - redis.expire(RedisKeys.alive(COUPON, 1, 0, "m3"), Duration.ofSeconds(2)).block(WAIT); + redis.opsForZSet().add(ALIVE, "m3", Long.parseLong(NOW) - 100).block(WAIT); enqueue("m3", "30", "0"); - assertThat(redis.getExpire(RedisKeys.alive(COUPON, 1, 0, "m3")).block(WAIT)) - .isGreaterThan(Duration.ofSeconds(20)); + assertThat(redis.opsForZSet().score(ALIVE, "m3").block(WAIT)) + .isEqualTo(Long.parseLong(NOW) + 30); } @Test @@ -135,6 +140,6 @@ private List enqueue(String memberId, String aliveTtl, String cap) { assertThat(redis.hasKey(MAX_SCORE).block(WAIT)).isFalse(); // alive 만 생기는 회귀를 잡는다. 검증이 첫 쓰기 앞이라는 계약은 // 세 키 전부에 걸린다 — 하나라도 새면 계약이 아니다. - assertThat(redis.hasKey(RedisKeys.alive(COUPON, 1, 0, "m1")).block(WAIT)).isFalse(); + assertThat(redis.hasKey(RedisKeys.alive(COUPON, 1, 0)).block(WAIT)).isFalse(); } } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/GraceReentryTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/GraceReentryTest.java index 20acf800..4e6e1f75 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/GraceReentryTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/GraceReentryTest.java @@ -25,6 +25,8 @@ @SpringBootTest class GraceReentryTest extends RedisContainerSupport { + private static final String NOW = "1800000000"; + private static final Duration WAIT = Duration.ofSeconds(5); private static final String COUPON = "grace"; private static final String QUEUE = RedisKeys.queue(COUPON, 1, 0); @@ -44,8 +46,8 @@ class GraceReentryTest extends RedisContainerSupport { private void enqueue(String memberId) { redis.execute(enqueueScript, - List.of(QUEUE, MAX_SCORE, RedisKeys.alive(COUPON, 1, 0, memberId)), - List.of(memberId, "86400", "30", "0")) + List.of(QUEUE, MAX_SCORE, RedisKeys.alive(COUPON, 1, 0)), + List.of(memberId, "86400", "30", "0", NOW)) .blockFirst(WAIT); } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/HardcodedKeyTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/HardcodedKeyTest.java new file mode 100644 index 00000000..d421a6bb --- /dev/null +++ b/src/test/java/com/kafkick/waiting/adapter/redis/HardcodedKeyTest.java @@ -0,0 +1,243 @@ +package com.kafkick.waiting.adapter.redis; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Pattern; +import java.util.stream.Stream; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +/** + * 키 문자열은 {@link RedisKeys} 에서만 만든다 (RD-3 · PK-R1). + * + *

두 곳에서 만들어지면 샤딩을 도입할 때 한쪽만 고쳐지고, 그때는 + * 진행 중인 큐가 통째로 유실된다. + */ +class HardcodedKeyTest { + + /** + * 어댑터만 본다. 도메인은 레디스를 모르므로(DS-1) 거기 있는 + * {@code "coupon:"} 같은 문자열은 리미터 예산 키지 레디스 키가 아니다. + * 그것까지 잡으면 오탐이고, 오탐이 나면 사람은 검사를 우회한다. + */ + private static final Path ADAPTER = Path.of("src/main/java/com/kafkick/waiting/adapter"); + private static final Path KEYS_CLASS = + Path.of("src/main/java/com/kafkick/waiting/adapter/redis/RedisKeys.java"); + + /** 이 저장소가 쓰는 키 접두사. 새 키를 만들면 여기도 함께 는다. */ + private static final Pattern KEY_LITERAL = Pattern.compile( + "\"(queue|maxscore|admitted|grace|alive|stock|gw|scheduler|coupons|coupon):"); + + private List violationsIn(Path root) throws IOException { + List violations = new ArrayList<>(); + try (Stream paths = Files.walk(root)) { + for (Path file : paths.filter(p -> p.toString().endsWith(".java")).toList()) { + if (file.equals(KEYS_CLASS)) { + continue; // 여기가 유일하게 허용되는 자리다 + } + // 주석 속 예시는 위반이 아니다. 오탐이 나면 사람은 + // 검사를 고치는 대신 우회한다. + String source = 주석을_지운다( + Files.readString(file, StandardCharsets.UTF_8)); + String[] lines = source.split("\n", -1); + for (int i = 0; i < lines.length; i++) { + if (KEY_LITERAL.matcher(lines[i]).find()) { + violations.add("%s:%d".formatted(file.getFileName(), i + 1)); + } + } + } + } + return violations; + } + + /** + * 주석만 지우고 문자열은 남긴다. 정규식으로는 못 한다 — 리터럴에 담긴 + * 주석 기호가 진짜 주석의 시작으로 잡히면 검사가 눈이 먼다. + * + *

지울 자리는 공백으로 덮어 줄 번호를 그대로 둔다. 통째로 + * 지우면 뒤따르는 위반이 엉뚱한 줄로 보고돼 엉뚱한 곳을 보게 된다. + */ + private static String 주석을_지운다(String source) { + char[] out = source.toCharArray(); + int n = out.length; + int i = 0; + while (i < n) { + char c = out[i]; + if (c == '/' && i + 1 < n && out[i + 1] == '/') { + while (i < n && out[i] != '\n') { + out[i++] = ' '; + } + } else if (c == '/' && i + 1 < n && out[i + 1] == '*') { + out[i++] = ' '; + out[i++] = ' '; + // 닫히지 않은 주석은 파일 끝까지가 주석이다 + while (i < n && !(out[i] == '*' && i + 1 < n && out[i + 1] == '/')) { + if (out[i] != '\n') { + out[i] = ' '; + } + i++; + } + i = Math.min(i + 2, n); + } else if (c == '"' && i + 2 < n && out[i + 1] == '"' && out[i + 2] == '"') { + i += 3; // 텍스트 블록. 안의 주석 기호는 코드가 아니다 + while (i < n && !(out[i] == '"' && i + 2 < n + && out[i + 1] == '"' && out[i + 2] == '"')) { + i += out[i] == '\\' ? 2 : 1; + } + i = Math.min(i + 3, n); + } else if (c == '"' || c == '\'') { + i++; // 리터럴 안은 건드리지 않는다. 키를 찾는 곳이 여기다 + while (i < n && out[i] != c && out[i] != '\n') { + i += out[i] == '\\' ? 2 : 1; + } + i++; + } else { + i++; + } + } + return new String(out); + } + + @Test + @DisplayName("키_문자열_리터럴이_RedisKeys_밖에_없다") + void 키_문자열_리터럴이_RedisKeys_밖에_없다() throws IOException { + List violations = violationsIn(ADAPTER); + + assertThat(violations) + .withFailMessage("RedisKeys 밖의 키 리터럴 %d 건%n%s", + violations.size(), String.join("\n", violations)) + .isEmpty(); + } + + @Test + @DisplayName("검사가_실제로_문다") + void 검사가_실제로_문다() throws IOException { + // 통과만 하는 검사는 모든 코드를 통과시킨다 (TS-9). + Path dir = Files.createTempDirectory("probe"); + Path probe = dir.resolve("Leak.java"); + Files.writeString(probe, + "class Leak {\n String key = \"queue:{c1}\";\n}\n", StandardCharsets.UTF_8); + + try { + // 무엇을 잡았는지까지 본다 — 엉뚱한 것을 잡아도 통과하면 안 된다. + assertThat(violationsIn(dir)).containsExactly("Leak.java:2"); + } finally { + Files.deleteIfExists(probe); + Files.deleteIfExists(dir); + } + } + + @Test + @DisplayName("블록_주석_속_예시는_위반이_아니다") + void 블록_주석_속_예시는_위반이_아니다() throws IOException { + Path dir = Files.createTempDirectory("probe"); + Path probe = dir.resolve("Doc.java"); + Files.writeString(probe, + "/**\n * 예: \"queue:{c1}\" 처럼 쓰면 안 된다\n */\nclass Doc {\n}\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(dir)).isEmpty(); + } finally { + Files.deleteIfExists(probe); + Files.deleteIfExists(dir); + } + } + + @Test + @DisplayName("주석_속_예시는_위반이_아니다") + void 주석_속_예시는_위반이_아니다() throws IOException { + Path dir = Files.createTempDirectory("probe"); + Path probe = dir.resolve("Fine.java"); + Files.writeString(probe, + "class Fine {\n // 예: \"queue:{c1}\" 처럼 쓰면 안 된다\n" + + " String key = RedisKeys.queue(\"c1\", 1, 0);\n}\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(dir)).isEmpty(); + } finally { + Files.deleteIfExists(probe); + Files.deleteIfExists(dir); + } + } + + @Test + @DisplayName("문자열_속_주석_기호가_검사를_눈멀게_하지_않는다") + void 문자열_속_주석_기호가_검사를_눈멀게_하지_않는다() throws IOException { + // 주석을 정규식으로 지우면 여기서 검사가 통째로 죽는다 — 리터럴 + // 안의 "/*" 가 주석의 시작으로 잡혀 뒤따르는 진짜 위반이 주석 + // 속으로 사라진다. **못 잡는데 통과하니 없는 것보다 나쁘다.** + Path dir = Files.createTempDirectory("probe"); + Path probe = dir.resolve("Blind.java"); + Files.writeString(probe, + "class Blind {\n" + + " String block = \"/*\";\n" + + " String line = \"//\";\n" + + " String key = \"queue:{c1}\";\n" + + " String close = \"*/\";\n}\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(dir)).containsExactly("Blind.java:4"); + } finally { + Files.deleteIfExists(probe); + Files.deleteIfExists(dir); + } + } + + @Test + @DisplayName("텍스트_블록_속_주석_기호도_코드가_아니다") + void 텍스트_블록_속_주석_기호도_코드가_아니다() throws IOException { + Path dir = Files.createTempDirectory("probe"); + Path probe = dir.resolve("Doc.java"); + Files.writeString(probe, + "class Doc {\n" + + " String sample = \"\"\"\n" + + " /* 여기는 문자열이다\n" + + " \"\"\";\n" + + " String key = \"queue:{c1}\";\n}\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(dir)).containsExactly("Doc.java:5"); + } finally { + Files.deleteIfExists(probe); + Files.deleteIfExists(dir); + } + } + + @Test + @DisplayName("도메인의_예산_키는_대상이_아니다") + void 도메인의_예산_키는_대상이_아니다() throws IOException { + // 도메인은 레디스를 모른다 (DS-1). 거기 있는 "coupon:" 은 리미터 + // 예산을 가르는 값이지 레디스 키가 아니다 — 잡으면 오탐이다. + assertThat(ADAPTER.toString()).endsWith("adapter"); + assertThat(violationsIn(ADAPTER)).isEmpty(); + } + + @Test + @DisplayName("블록_주석_뒤의_위반은_원본_줄_번호로_보고된다") + void 블록_주석_뒤의_위반은_원본_줄_번호로_보고된다() throws IOException { + // 주석을 통째로 지우면 줄 번호가 앞으로 밀려 엉뚱한 곳을 보게 된다. + Path dir = Files.createTempDirectory("probe"); + Path probe = dir.resolve("Shifted.java"); + Files.writeString(probe, + "/**\n * 여러\n * 줄\n * 주석\n */\nclass Shifted {\n" + + " String key = \"queue:{c1}\";\n}\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(dir)).containsExactly("Shifted.java:7"); + } finally { + Files.deleteIfExists(probe); + Files.deleteIfExists(dir); + } + } +} diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/HashTagSlotTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/HashTagSlotTest.java index dc81a7a3..69b0602b 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/HashTagSlotTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/HashTagSlotTest.java @@ -30,7 +30,7 @@ private int slotOf(String key) { RedisKeys.maxScore("c1", 4, 3), RedisKeys.admitted("c1", 4, 3), RedisKeys.grace("c1", 4, 3), - RedisKeys.alive("c1", 4, 3, "m9")); + RedisKeys.alive("c1", 4, 3)); assertThat(keys).extracting(this::slotOf).containsOnly(slotOf(keys.get(0))); } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/LuaKeysDeclarationTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/LuaKeysDeclarationTest.java new file mode 100644 index 00000000..9c70406d --- /dev/null +++ b/src/test/java/com/kafkick/waiting/adapter/redis/LuaKeysDeclarationTest.java @@ -0,0 +1,183 @@ +package com.kafkick.waiting.adapter.redis; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Stream; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +/** + * Lua 는 {@code KEYS} 에 선언되지 않은 키를 만지지 않는다 (RD-1 · G3.5). + * + *

클러스터에서 미선언 키 접근은 런타임 오류다. 여기서 못 잡으면 + * 부하 시험 중에 터진다 — 그때는 스크립트를 전부 다시 써야 한다. + */ +class LuaKeysDeclarationTest { + + private static final Path SCRIPTS = Path.of("src/main/resources/redis"); + + /** {@code redis.call('CMD', <키>` — 명령 뒤 첫 인자가 키 자리다. */ + private static final Pattern KEY_ARGUMENT = Pattern.compile( + "redis\\.(?:call|pcall)\\(\\s*'[A-Za-z]+'\\s*,\\s*([^,)]+)"); + + private List scripts() throws IOException { + try (Stream paths = Files.list(SCRIPTS)) { + return paths.filter(p -> p.toString().endsWith(".lua")).sorted().toList(); + } + } + + /** 주석을 걷어낸다 — 주석 속 예시가 위반으로 잡히면 검사를 안 믿게 된다. */ + private String codeOf(Path script) throws IOException { + return Files.readAllLines(script, StandardCharsets.UTF_8).stream() + .map(line -> line.replaceAll("--.*", "")) + .reduce("", (a, b) -> a + "\n" + b); + } + + /** + * {@code KEYS[n]} 에서 온 값만 키 자리에 올 수 있다. + * + *

지역 변수를 무조건 허용하면 {@code local rogue = 'queue:{c1}'} 를 + * 거쳐 들어오는 것을 못 막는다 — 검사가 있는데 아무것도 안 보는 상태다. + */ + private List violationsIn(Path script) throws IOException { + String code = codeOf(script); + + // KEYS[n] 이 대입된 지역 변수만 모은다. 문자열 조립은 안 받는다. + Set derived = new HashSet<>(); + Matcher assignment = Pattern + .compile("local\\s+([A-Za-z_][A-Za-z0-9_]*)\\s*=\\s*KEYS\\[") + .matcher(code); + while (assignment.find()) { + derived.add(assignment.group(1)); + } + + List violations = new ArrayList<>(); + Matcher matcher = KEY_ARGUMENT.matcher(code); + while (matcher.find()) { + String argument = matcher.group(1).strip(); + boolean declared = argument.startsWith("KEYS[") || derived.contains(argument); + if (!declared) { + violations.add("%s: %s".formatted(script.getFileName(), argument)); + } + } + return violations; + } + + @Test + @DisplayName("모든_Lua가_KEYS에_선언된_키만_만진다") + void 모든_Lua가_KEYS에_선언된_키만_만진다() throws IOException { + List violations = new ArrayList<>(); + for (Path script : scripts()) { + violations.addAll(violationsIn(script)); + } + + assertThat(violations) + .withFailMessage("KEYS 미선언 키 접근 %d 건%n%s", + violations.size(), String.join("\n", violations)) + .isEmpty(); + } + + @Test + @DisplayName("모든_Lua에_KEYS와_ARGV_계약_주석이_있다") + void 모든_Lua에_KEYS와_ARGV_계약_주석이_있다() throws IOException { + // 계약이 없으면 호출부가 인자 순서를 추측하게 된다 (RD-10). + // **상단 주석 블록만 본다** — 코드에서 KEYS[1] 을 쓰는 것과 + // 계약을 적어 둔 것은 다르다. + for (Path script : scripts()) { + String header = headerOf(script); + assertThat(header) + .withFailMessage("%s 상단에 KEYS 계약이 없다", script.getFileName()) + .contains("KEYS[1]"); + assertThat(header) + .withFailMessage("%s 상단에 ARGV 계약이 없다", script.getFileName()) + .contains("ARGV[1]"); + } + } + + /** 첫 코드 줄 전까지의 주석 블록. */ + private String headerOf(Path script) throws IOException { + StringBuilder header = new StringBuilder(); + for (String line : Files.readAllLines(script, StandardCharsets.UTF_8)) { + String trimmed = line.strip(); + if (!trimmed.isEmpty() && !trimmed.startsWith("--")) { + break; + } + header.append(line).append('\n'); + } + return header.toString(); + } + + @Test + @DisplayName("문자열_조립으로_만든_키는_위반이다") + void 문자열_조립으로_만든_키는_위반이다() throws IOException { + // 지역 변수를 무조건 허용하면 이 경로로 다 새어 나간다. + Path probe = Files.createTempFile("probe", ".lua"); + Files.writeString(probe, + "local rogue = 'queue:{c1}'\nredis.call('ZCARD', rogue)\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(probe)).singleElement().asString().contains("rogue"); + } finally { + Files.deleteIfExists(probe); + } + } + + @Test + @DisplayName("KEYS에서_온_지역_변수는_위반이_아니다") + void KEYS에서_온_지역_변수는_위반이_아니다() throws IOException { + Path probe = Files.createTempFile("probe", ".lua"); + Files.writeString(probe, + "local queue = KEYS[1]\nredis.call('ZCARD', queue)\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(probe)).isEmpty(); + } finally { + Files.deleteIfExists(probe); + } + } + + @Test + @DisplayName("검사가_실제로_문다") + void 검사가_실제로_문다() throws IOException { + // 통과만 하는 검사는 모든 스크립트를 통과시킨다 (TS-9). + Path probe = Files.createTempFile("probe", ".lua"); + Files.writeString(probe, "redis.call('ZCARD', 'queue:{c1}')\n", StandardCharsets.UTF_8); + + try { + // 무엇을 잡았는지까지 본다 — 엉뚱한 것을 잡아도 통과하면 안 된다. + assertThat(violationsIn(probe)) + .singleElement().asString().contains("'queue:{c1}'"); + } finally { + Files.deleteIfExists(probe); + } + } + + @Test + @DisplayName("주석_속_예시는_위반이_아니다") + void 주석_속_예시는_위반이_아니다() throws IOException { + // 오탐이 나면 사람은 검사를 고치는 대신 우회한다. + Path probe = Files.createTempFile("probe", ".lua"); + Files.writeString(probe, + "-- redis.call('ZCARD', 'queue:{c1}') 처럼 쓰면 안 된다\n" + + "redis.call('ZCARD', KEYS[1])\n", + StandardCharsets.UTF_8); + + try { + assertThat(violationsIn(probe)).isEmpty(); + } finally { + Files.deleteIfExists(probe); + } + } +} diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/QueueStatusTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/QueueStatusTest.java index f15b044d..ce4d1c93 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/QueueStatusTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/QueueStatusTest.java @@ -25,6 +25,8 @@ @SpringBootTest class QueueStatusTest extends RedisContainerSupport { + private static final String NOW = "1800000000"; + private static final Duration WAIT = Duration.ofSeconds(5); private static final String COUPON = "status"; private static final String QUEUE = RedisKeys.queue(COUPON, 1, 0); @@ -50,12 +52,12 @@ class QueueStatusTest extends RedisContainerSupport { } private String alive(String memberId) { - return RedisKeys.alive(COUPON, 1, 0, memberId); + return RedisKeys.alive(COUPON, 1, 0); } private void enqueue(String memberId) { redis.execute(enqueueScript, List.of(QUEUE, MAX_SCORE, alive(memberId)), - List.of(memberId, "86400", ALIVE_TTL, "0")) + List.of(memberId, "86400", ALIVE_TTL, "0", NOW)) .blockFirst(WAIT); } @@ -64,7 +66,7 @@ private List status(String memberId) { return (List) redis.execute( statusScript, List.of(QUEUE, ADMITTED, alive(memberId), GRACE), - List.of(memberId, ALIVE_TTL)) + List.of(memberId, ALIVE_TTL, NOW)) .blockFirst(WAIT); } @@ -105,12 +107,13 @@ private long rank(List r) { // **경계에 붙이지 않는다.** 2초로 줄여 두면 CI 지연 한 번에 키가 // 먼저 사라져 갱신이 아예 안 돌고, 그러면 이 시험이 불안정해진다. enqueue("m0"); - redis.expire(alive("m0"), Duration.ofSeconds(10)).block(WAIT); + redis.opsForZSet().add(alive("m0"), "m0", Long.parseLong(NOW) - 100).block(WAIT); assertThat(state(status("m0"))).isEqualTo("WAITING"); - assertThat(redis.getExpire(alive("m0")).block(WAIT)) - .isGreaterThan(Duration.ofSeconds(20)); + // score 가 만료 시각이다. 조회 한 번에 미래로 밀려야 한다. + assertThat(redis.opsForZSet().score(alive("m0"), "m0").block(WAIT)) + .isEqualTo(Long.parseLong(NOW) + 30); } @Test @@ -198,7 +201,7 @@ private long rank(List r) { assertThatThrownBy(() -> redis.execute(statusScript, List.of(QUEUE, ADMITTED, alive("m0"), GRACE), - List.of("m0", "0")) + List.of("m0", "0", NOW)) .blockFirst(WAIT)) .rootCause() .hasMessageContaining("alive TTL"); @@ -206,7 +209,7 @@ private long rank(List r) { // 쓰기 대상 전부를 본다. 하나만 보면 ZREM 이나 HSET 이 먼저 // 일어난 회귀를 통과시킨다. 순번은 **그대로**여야 한다 — // 있기만 하면 되는 게 아니라 값이 안 바뀌어야 한다. - assertThat(redis.hasKey(alive("m0")).block(WAIT)).isFalse(); + assertThat(redis.opsForZSet().score(alive("m0"), "m0").block(WAIT)).isNull(); assertThat(redis.opsForZSet().score(QUEUE, "m0").block(WAIT)).isEqualTo(before); assertThat(redis.opsForHash().hasKey(GRACE, "m0").block(WAIT)).isFalse(); } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/RankMonotonicityIntegrationTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/RankMonotonicityIntegrationTest.java index eb0f3744..dbc36f55 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/RankMonotonicityIntegrationTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/RankMonotonicityIntegrationTest.java @@ -31,6 +31,8 @@ @SpringBootTest class RankMonotonicityIntegrationTest extends RedisContainerSupport { + private static final String NOW = "1800000000"; + private static final long SEED = 20260819L; private static final int OPERATIONS = 3_000; private static final int PEOPLE = 200; @@ -53,8 +55,8 @@ class RankMonotonicityIntegrationTest extends RedisContainerSupport { private void enqueue(String memberId) { redis.execute(enqueueScript, - List.of(QUEUE, MAX_SCORE, RedisKeys.alive(COUPON, 1, 0, memberId)), - List.of(memberId, "86400", "30", "0")) + List.of(QUEUE, MAX_SCORE, RedisKeys.alive(COUPON, 1, 0)), + List.of(memberId, "86400", "30", "0", NOW)) .blockFirst(WAIT); } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/RedisKeysTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/RedisKeysTest.java index d2e45cdb..b80547ef 100644 --- a/src/test/java/com/kafkick/waiting/adapter/redis/RedisKeysTest.java +++ b/src/test/java/com/kafkick/waiting/adapter/redis/RedisKeysTest.java @@ -35,7 +35,7 @@ class RedisKeysTest { assertThat(RedisKeys.maxScore("c1", 1, 0)).isEqualTo("maxscore:{c1}"); assertThat(RedisKeys.admitted("c1", 1, 0)).isEqualTo("admitted:{c1}"); assertThat(RedisKeys.grace("c1", 1, 0)).isEqualTo("grace:{c1}"); - assertThat(RedisKeys.alive("c1", 1, 0, "m9")).isEqualTo("alive:{c1}:m9"); + assertThat(RedisKeys.alive("c1", 1, 0)).isEqualTo("alive:{c1}"); } @Test @@ -43,7 +43,7 @@ class RedisKeysTest { void 샤드가_여럿이면_접미사가_붙는다() { assertThat(RedisKeys.queue("c1", 4, 3)).isEqualTo("queue:{c1:3}"); assertThat(RedisKeys.maxScore("c1", 4, 3)).isEqualTo("maxscore:{c1:3}"); - assertThat(RedisKeys.alive("c1", 4, 3, "m9")).isEqualTo("alive:{c1:3}:m9"); + assertThat(RedisKeys.alive("c1", 4, 3)).isEqualTo("alive:{c1:3}"); } @Test @@ -55,7 +55,7 @@ class RedisKeysTest { RedisKeys.maxScore("c1", 4, 3), RedisKeys.admitted("c1", 4, 3), RedisKeys.grace("c1", 4, 3), - RedisKeys.alive("c1", 4, 3, "m9")); + RedisKeys.alive("c1", 4, 3)); assertThat(keys).allMatch(k -> RedisKeys.hashTagOf(k).equals("c1:3")); } @@ -76,7 +76,9 @@ class RedisKeysTest { .isInstanceOf(IllegalArgumentException.class); assertThatThrownBy(() -> RedisKeys.queue("c:1", 1, 0)) .isInstanceOf(IllegalArgumentException.class); - assertThatThrownBy(() -> RedisKeys.alive("c1", 1, 0, "m{9}")) + assertThatThrownBy(() -> RedisKeys.alive("c{1}", 1, 0)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> RedisKeys.stock("c:1")) .isInstanceOf(IllegalArgumentException.class); } diff --git a/src/test/java/com/kafkick/waiting/adapter/redis/SweepTest.java b/src/test/java/com/kafkick/waiting/adapter/redis/SweepTest.java new file mode 100644 index 00000000..5af8ba83 --- /dev/null +++ b/src/test/java/com/kafkick/waiting/adapter/redis/SweepTest.java @@ -0,0 +1,271 @@ +package com.kafkick.waiting.adapter.redis; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.Duration; +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.core.io.ClassPathResource; +import org.springframework.data.redis.core.ReactiveStringRedisTemplate; +import org.springframework.data.redis.core.script.RedisScript; + +/** + * 이탈자 청소. + * + *

앞부분만 훑는다. 2만 명 큐에서 전체를 보면 청소 자체가 부하다. + * 뒤엣사람은 아직 폴링할 차례가 안 왔을 뿐 죽은 것이 아니다. + */ +@Tag("integration") +@SpringBootTest +class SweepTest extends RedisContainerSupport { + + private static final Duration WAIT = Duration.ofSeconds(5); + private static final String COUPON = "sweep"; + private static final String QUEUE = RedisKeys.queue(COUPON, 1, 0); + private static final String MAX_SCORE = RedisKeys.maxScore(COUPON, 1, 0); + private static final String GRACE = RedisKeys.grace(COUPON, 1, 0); + private static final String ALIVE = RedisKeys.alive(COUPON, 1, 0); + + /** 시각을 주입한다 — 실제 시계에 기대면 만료 시험이 흔들린다 (TS-4). */ + private static final long NOW = 1_800_000_000L; + private static final String RETENTION = "300"; + private static final String BUDGET = "1000"; + + @Autowired + private ReactiveStringRedisTemplate redis; + + private RedisScript enqueueScript; + private RedisScript sweepScript; + + @BeforeEach + void 준비() { + enqueueScript = RedisScript.of(new ClassPathResource("redis/enqueue.lua"), List.class); + sweepScript = RedisScript.of(new ClassPathResource("redis/sweep.lua"), List.class); + redis.delete(QUEUE, MAX_SCORE, GRACE, ALIVE).block(WAIT); + } + + private void enqueue(String memberId) { + redis.execute(enqueueScript, + List.of(QUEUE, MAX_SCORE, ALIVE), + List.of(memberId, "86400", "3600", "0", String.valueOf(NOW))) + .blockFirst(WAIT); + } + + @SuppressWarnings("unchecked") + private List sweep(String limit, String budget, String cursor) { + return (List) redis.execute( + sweepScript, + List.of(QUEUE, GRACE, ALIVE), + List.of(limit, String.valueOf(NOW), RETENTION, budget, cursor)) + .blockFirst(WAIT); + } + + private List sweep(String limit) { + return sweep(limit, BUDGET, "0"); + } + + private String nextCursor(List r) { + return String.valueOf(r.get(3)); + } + + private long swept(List r) { + return Long.parseLong(String.valueOf(r.get(0))); + } + + /** 유예 기록 정리 수. 반환값 두 번째는 생존 신호 정리 수다. */ + private long expired(List r) { + return Long.parseLong(String.valueOf(r.get(2))); + } + + @Test + @DisplayName("생존_키가_없는_앞부분_항목이_제거된다") + void 생존_키가_없는_앞부분_항목이_제거된다() { + enqueue("m0"); + enqueue("m1"); + enqueue("m2"); + double keptScore = redis.opsForZSet().score(QUEUE, "m2").block(WAIT); + redis.opsForZSet().remove(ALIVE, "m0", "m1").block(WAIT); + + assertThat(swept(sweep("10"))).isEqualTo(2); + + // 크기만 보면 엉뚱한 사람을 지운 구현도 통과한다. 누가 빠지고 + // 누가 남았는지를 직접 본다. + assertThat(redis.opsForZSet().score(QUEUE, "m0").block(WAIT)).isNull(); + assertThat(redis.opsForZSet().score(QUEUE, "m1").block(WAIT)).isNull(); + assertThat(redis.opsForZSet().score(QUEUE, "m2").block(WAIT)).isEqualTo(keptScore); + } + + @Test + @DisplayName("생존_키가_있으면_건드리지_않는다") + void 생존_키가_있으면_건드리지_않는다() { + enqueue("m0"); + enqueue("m1"); + + assertThat(swept(sweep("10"))).isZero(); + assertThat(redis.opsForZSet().size(QUEUE).block(WAIT)).isEqualTo(2); + } + + @Test + @DisplayName("검사_범위_밖은_보지_않는다") + void 검사_범위_밖은_보지_않는다() { + for (int i = 0; i < 5; i++) { + enqueue("m" + i); + redis.opsForZSet().remove(ALIVE, "m" + i).block(WAIT); + } + + double kept2 = redis.opsForZSet().score(QUEUE, "m2").block(WAIT); + double kept4 = redis.opsForZSet().score(QUEUE, "m4").block(WAIT); + + assertThat(swept(sweep("2"))).isEqualTo(2); + + // 앞 둘만 빠지고 범위 밖은 **순번까지 그대로** 남는다 + assertThat(redis.opsForZSet().score(QUEUE, "m0").block(WAIT)).isNull(); + assertThat(redis.opsForZSet().score(QUEUE, "m1").block(WAIT)).isNull(); + assertThat(redis.opsForZSet().score(QUEUE, "m2").block(WAIT)).isEqualTo(kept2); + assertThat(redis.opsForZSet().score(QUEUE, "m4").block(WAIT)).isEqualTo(kept4); + } + + @Test + @DisplayName("검사_범위가_인자로_주어진다") + void 검사_범위가_인자로_주어진다() { + // 부하와 정확도의 맞바꿈이라 배포 없이 조절할 수 있어야 한다 (P-1). + for (int i = 0; i < 5; i++) { + enqueue("m" + i); + redis.opsForZSet().remove(ALIVE, "m" + i).block(WAIT); + } + + assertThat(swept(sweep("1"))).isOne(); + assertThat(swept(sweep("4"))).isEqualTo(4); + } + + @Test + @DisplayName("제거된_사람이_유예_기록에_남는다") + void 제거된_사람이_유예_기록에_남는다() { + // 제거와 기록이 갈리면 자리도 잃고 재방문자로도 식별 안 되는 + // 사람이 생긴다. 같은 스크립트 안에서 한다. + enqueue("m0"); + redis.opsForZSet().remove(ALIVE, "m0").block(WAIT); + + sweep("10"); + + assertThat(redis.opsForHash().get(GRACE, "m0").block(WAIT)) + .isEqualTo(String.valueOf(NOW)); + } + + @Test + @DisplayName("만료된_유예_기록이_정리된다") + void 만료된_유예_기록이_정리된다() { + redis.opsForHash().put(GRACE, "old", String.valueOf(NOW - 400)).block(WAIT); + redis.opsForHash().put(GRACE, "fresh", String.valueOf(NOW - 100)).block(WAIT); + + assertThat(expired(sweep("10"))).isOne(); + assertThat(redis.opsForHash().hasKey(GRACE, "old").block(WAIT)).isFalse(); + assertThat(redis.opsForHash().hasKey(GRACE, "fresh").block(WAIT)).isTrue(); + } + + @Test + @DisplayName("유예_기록이_무한히_쌓이지_않는다") + void 유예_기록이_무한히_쌓이지_않는다() { + // 만료가 없으면 이 해시가 영원히 자란다 (RD-7). + for (int i = 0; i < 50; i++) { + redis.opsForHash().put(GRACE, "old" + i, String.valueOf(NOW - 1000)).block(WAIT); + } + + sweep("10"); + + assertThat(redis.opsForHash().size(GRACE).block(WAIT)).isZero(); + } + + @Test + @DisplayName("값이_깨진_유예_기록도_정리된다") + void 값이_깨진_유예_기록도_정리된다() { + // 숫자가 아니면 언제 것인지 알 수 없다. 남겨 두면 영원히 안 지워진다. + redis.opsForHash().put(GRACE, "broken", "언제인지모름").block(WAIT); + + assertThat(expired(sweep("10"))).isOne(); + assertThat(redis.opsForHash().hasKey(GRACE, "broken").block(WAIT)).isFalse(); + } + + @Test + @DisplayName("잘못된_인자는_아무것도_바꾸지_않는다") + void 잘못된_인자는_아무것도_바꾸지_않는다() { + enqueue("m0"); + double before = redis.opsForZSet().score(QUEUE, "m0").block(WAIT); + redis.opsForZSet().remove(ALIVE, "m0").block(WAIT); + + assertThatThrownBy(() -> sweep("0")).rootCause().hasMessageContaining("검사 범위"); + + assertThat(redis.opsForZSet().score(QUEUE, "m0").block(WAIT)).isEqualTo(before); + assertThat(redis.opsForHash().size(GRACE).block(WAIT)).isZero(); + } + + @Test + @DisplayName("정리_예산이_한_번의_실행을_묶는다") + void 정리_예산이_한_번의_실행을_묶는다() { + // COUNT 는 힌트지 상한이 아니다. 받은 것 중 예산만큼만 지워야 + // 한 번의 실행이 유계다. + for (int i = 0; i < 40; i++) { + redis.opsForHash().put(GRACE, "old" + i, String.valueOf(NOW - 1000)).block(WAIT); + } + + assertThat(expired(sweep("10", "5", "0"))).isEqualTo(5); + assertThat(redis.opsForHash().size(GRACE).block(WAIT)).isEqualTo(35); + } + + @Test + @DisplayName("커서를_이어_넘기면_전부_정리된다") + void 커서를_이어_넘기면_전부_정리된다() { + // 한 번에 다 안 지우는 대신 다음 틱이 이어받는다. 커서가 돌지 + // 않으면 같은 앞부분만 계속 보고 뒤는 영영 안 지워진다. + for (int i = 0; i < 40; i++) { + redis.opsForHash().put(GRACE, "old" + i, String.valueOf(NOW - 1000)).block(WAIT); + } + + String cursor = "0"; + for (int round = 0; round < 20; round++) { + cursor = nextCursor(sweep("10", "5", cursor)); + if (Boolean.TRUE.equals(redis.opsForHash().size(GRACE).block(WAIT) == 0L)) { + break; + } + } + + assertThat(redis.opsForHash().size(GRACE).block(WAIT)).isZero(); + } + + @Test + @DisplayName("잘못된_커서는_아무것도_바꾸지_않는다") + void 잘못된_커서는_아무것도_바꾸지_않는다() { + // 커서 검증이 쓰기 뒤에 있으면 앞의 쓰기가 남는다 — Lua 는 + // 롤백하지 않는다. + enqueue("m0"); + double before = redis.opsForZSet().score(QUEUE, "m0").block(WAIT); + redis.opsForZSet().remove(ALIVE, "m0").block(WAIT); + redis.opsForHash().put(GRACE, "keep", String.valueOf(NOW)).block(WAIT); + + assertThatThrownBy(() -> sweep("10", "5", "abc")).rootCause() + .hasMessageContaining("커서"); + + assertThat(redis.opsForZSet().score(QUEUE, "m0").block(WAIT)).isEqualTo(before); + assertThat(redis.opsForHash().hasKey(GRACE, "keep").block(WAIT)).isTrue(); + } + + @Test + @DisplayName("만료된_생존_신호도_예산_안에서_걷는다") + void 만료된_생존_신호도_예산_안에서_걷는다() { + // 한 번에 다 지우려 하면 그 자체가 오래 걸린다. + for (int i = 0; i < 20; i++) { + redis.opsForZSet().add(ALIVE, "gone" + i, NOW - 100).block(WAIT); + } + + List result = sweep("1", "5", "0"); + + assertThat(Long.parseLong(String.valueOf(result.get(1)))).isEqualTo(5); + assertThat(redis.opsForZSet().size(ALIVE).block(WAIT)).isEqualTo(15); + } +}