[FEAT] 데이터 차감 Redis-Only 전면 전환 및 속도 제한(QoS) 로직 개편#4
Merged
Conversation
…alDateTime 으로 변환하고, done log에 작성
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. 🗂️ Base branches to auto review (2)
Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughRedis Streams 트리밍 설정 추가, Lua 차감 스크립트에 finishedAtEpochMillis·속도 예약·timeout 로직 도입, QoS·앱 속도 예약 키 생성/전달 확장, 정책 CAS와 ACK→XDEL 흐름 개선, 관련 계약·설정·테스트 보강. ChangesQoS/앱 속도 제한 예약 및 타이밍 시스템
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested labels
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
✨ Simplify code
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Owner
Author
|
@CodeRabbit review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
src/main/java/com/pooli/traffic/service/decision/TrafficDeductOrchestratorService.java (1)
52-52:⚠️ Potential issue | 🟡 Minor | ⚡ Quick win
startedAt타임존 불일치 문제
startedAt은LocalDateTime.now()로 시스템 기본 타임존을 사용하지만,finishedAt은trafficRedisRuntimePolicy.zoneId()를 사용합니다. 시스템 기본 타임존과 설정된 타임존이 다를 경우,startedAt과finishedAt간 시간 비교나 레이턴시 계산에서 오류가 발생할 수 있습니다.🔧 타임존 일관성 수정 제안
- LocalDateTime startedAt = LocalDateTime.now(); + LocalDateTime startedAt = LocalDateTime.now(trafficRedisRuntimePolicy.zoneId());🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/java/com/pooli/traffic/service/decision/TrafficDeductOrchestratorService.java` at line 52, startedAt is created with LocalDateTime.now() using the system default zone while finishedAt uses trafficRedisRuntimePolicy.zoneId(), causing timezone inconsistency; change startedAt to be created with the same zone as finishedAt (use trafficRedisRuntimePolicy.zoneId()) so both timestamps are in the same timezone (e.g., LocalDateTime.now(trafficRedisRuntimePolicy.zoneId()) or switch both to ZonedDateTime.now(trafficRedisRuntimePolicy.zoneId())), and update any downstream latency/calculation logic in TrafficDeductOrchestratorService that references startedAt/finishedAt accordingly.src/main/java/com/pooli/traffic/service/outbox/TrafficPolicyVersionedRedisService.java (1)
86-112: 🧹 Nitpick | 🔵 Trivial | 💤 Low value
ArrayList초기 용량 지정으로 재할당 방지.
qosSpeedLimitNextAvailableKeys의 크기를 알 수 있으므로,ArrayList생성 시 초기 용량을 지정하면 내부 배열 재할당을 피할 수 있습니다.♻️ 초기 용량 지정 제안
- List<String> keys = new ArrayList<>(); + int initialCapacity = 3 + (qosSpeedLimitNextAvailableKeys == null ? 0 : qosSpeedLimitNextAvailableKeys.size()); + List<String> keys = new ArrayList<>(initialCapacity); keys.add(appDataKey); keys.add(appSpeedKey); keys.add(appWhitelistKey);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/java/com/pooli/traffic/service/outbox/TrafficPolicyVersionedRedisService.java` around lines 86 - 112, In syncAppPolicySnapshot in TrafficPolicyVersionedRedisService the keys List is created with new ArrayList<>() causing potential reallocations; compute an initial capacity (e.g., 3 + (qosSpeedLimitNextAvailableKeys == null ? 0 : qosSpeedLimitNextAvailableKeys.size())) and construct the ArrayList with that capacity before adding appDataKey, appSpeedKey, appWhitelistKey and qosSpeedLimitNextAvailableKeys to avoid internal array growth.src/main/java/com/pooli/traffic/service/policy/TrafficPolicyWriteThroughService.java (1)
319-373: 🧹 Nitpick | 🔵 Trivial | 💤 Low value
LinkedHashSet대신HashSet사용 가능.
appIdsToClearReservation의 순서가 DEL 작업에 영향을 주지 않으므로, 삽입 순서를 유지하는LinkedHashSet대신 일반HashSet을 사용해도 됩니다. 성능 차이는 미미하지만 의도를 더 명확히 표현합니다.♻️ HashSet 사용 제안
- Set<Integer> appIdsToClearReservation = new LinkedHashSet<>(); + Set<Integer> appIdsToClearReservation = new HashSet<>();그리고 사용하지 않는
LinkedHashSetimport도 제거할 수 있습니다:-import java.util.LinkedHashSet;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/java/com/pooli/traffic/service/policy/TrafficPolicyWriteThroughService.java` around lines 319 - 373, Replace the LinkedHashSet used for appIdsToClearReservation in syncAppPolicySnapshotUntracked with a plain HashSet since iteration order isn't required; update the declaration of appIdsToClearReservation to use HashSet<Integer> and leave the rest of the logic that builds qosSpeedLimitNextAvailableKeys unchanged, and remove any unused LinkedHashSet import from the file.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/redis-only-release-checklist.md`:
- Line 20: 문서의 하드코딩된 Redis 버전("M5 확인 결과 ... redis_version=7.4.2") 때문에 체크리스트가 쉽게
stale 해질 수 있으므로 해당 하드코딩된 버전 문자열을 제거하고 대신 검증 절차와 결과를 어디에 기록할지 명시하세요; 예를 들어 `M5 확인
결과` 항목에는 "`.env` 기준 cache/streams Redis의 redis_version을 확인" 같은 절차만 남기고, 실제 버전값은
릴리즈 노트나 PR 코멘트(릴리즈 기록 위치)에 기록하도록 안내문을 추가하여 `redis_version` 값은 문서에 남기지 않도록 수정하세요.
In
`@src/main/java/com/pooli/traffic/service/invoke/TrafficStreamInfraService.java`:
- Around line 171-175: The early-return branch for acknowledgedCount <= 0 (in
TrafficStreamInfraService where acknowledgedCount,
log.warn("traffic_stream_ack_zero", streamKey, group, recordId.getValue()) and
the subsequent skipping of XDEL occur) must be instrumented and surfaced: add a
metric/alert increment when this branch is hit, enrich the log with diagnostic
context (consumer group name, pending count, stream info and relevant IDs), and
ensure any retry/duplicate-ACK logic is audited so operations can investigate;
keep the current return behavior but emit the metric and richer log so frequent
occurrences trigger investigation.
- Around line 158-179: The current flow performs XACK then XDEL per record
causing two Redis round-trips; change to batch/pipeline these operations: add a
new method (e.g., deleteAcknowledgedRecordsBatch or
deleteAcknowledgedRecordPipeline) that accepts a collection of recordIds and
uses Redis pipelining or streamOps().executePipelined to send all
acknowledge(...) calls and then a single batched XDEL (or pipelined XDELs) for
the succeeded ids; update the caller that currently invokes
executeStreamsOperation/streamOps().acknowledge and deleteAcknowledgedRecord to
collect recordIds per poll and call the new batch/pipeline method so multiple
records are ACKed and XDELed in a single network round-trip while preserving
best-effort semantics (ignore XDEL failures).
- Around line 182-197: Add metrics instrumentation to deleteAcknowledgedRecord
to count XDEL failures and zero-delete cases: inside the try block after
computing deletedCount (in deleteAcknowledgedRecord using
executeStreamsOperation and streamOps().delete) increment a suitable counter
(e.g., trafficRedisAvailabilityMetrics.incrementXdelZeroCount() or a new metric
like incrementXdelZeroCount) when deletedCount <= 0L, and inside the
catch(RuntimeException e) increment a failure counter (e.g.,
trafficRedisAvailabilityMetrics.incrementXdelFailureCount()) before logging;
ensure trafficRedisAvailabilityMetrics is injected/available and choose stable
metric names that distinguish zero-deletes from exception-failures.
In `@src/main/resources/lua/traffic/app_policy_snapshot_cas.lua`:
- Around line 14-19: 현재 루프에서 각 키마다 개별적으로 호출하는 redis.call('DEL', ...)를 하나의 배치
DEL로 합쳐 오버헤드를 줄이세요: KEYS 배열을 순회하면서 KEYS[1..#KEYS] 중 nil 또는 빈 문자열이 아닌 값만 수집해
valid_keys 테이블/배열로 만들고, 마지막에 redis.call('DEL', table.unpack(valid_keys)) (또는 Lua
환경에 맞는 unpack) 한 번만 호출하도록 변경하세요; 관련 식별자: KEYS, KEYS[i], redis.call('DEL', ...).
In `@src/main/resources/lua/traffic/deduct_unified.lua`:
- Around line 352-388: The reservation key qos_speed_limit_next_available_key is
written with redis.call("SET", qos_speed_limit_next_available_key,
finished_at_epoch_millis) but no TTL is set, which can leave stale keys
indefinitely; change the write to set an expiration (e.g., 86400 seconds) by
using Redis TTL-aware command (SET with EX or SETEX) when updating
qos_speed_limit_next_available_key (or introduce a constant like
SPEED_LIMIT_TTL_SECONDS) so the key expires if the line+app is unused; keep the
same value logic (finished_at_epoch_millis) and ensure the new TTL is applied
whenever reservation is updated in this code path.
In `@src/test/java/com/pooli/common/config/AppStreamsPropertiesTest.java`:
- Around line 69-76: Update the test in AppStreamsPropertiesTest to cover both
zero and negative boundary values: call
AppStreamsProperties#setTrafficRequestMaxLength with 0 and with a negative value
(e.g., -1) and assert that AppStreamsProperties#requireTrafficRequestMaxLength
throws TrafficStreamBootstrapException for each; convert the current single-case
test into a parameterized test (or add a second test) that iterates the invalid
inputs so both 0 and negative inputs are verified.
In
`@src/test/java/com/pooli/traffic/acceptance/TrafficDataPolicyAcceptanceTest.java`:
- Around line 198-200: Replace the hardcoded expectedDurationMs with a
calculated value derived from requestBytes and speedBytesPerSecond: compute the
transfer time in milliseconds using requestBytes and speedBytesPerSecond (e.g.,
ceil((double)requestBytes / speedBytesPerSecond * 1000) or integer math like
(requestBytes * 1000 + speedBytesPerSecond - 1) / speedBytesPerSecond) and
assign that result to expectedDurationMs so the test stays correct when
requestBytes or speedBytesPerSecond change; update the code that defines
expectedDurationMs accordingly (referencing the variables requestBytes and
speedBytesPerSecond).
In
`@src/test/java/com/pooli/traffic/service/invoke/TrafficRequestEnqueueServiceTest.java`:
- Line 86: The current assertion uses a brittle substring check on ((Map<?,?>)
record.getValue()).get("payload"); instead parse the payload JSON and validate
the traceId field directly: extract payload from record.getValue(), parse it
(e.g., with your test ObjectMapper) into a Map or DTO, then assert the traceId
field exists and equals or matches the expected value/pattern. Update the
assertion in TrafficRequestEnqueueServiceTest to use the parsed payload's
"traceId" instead of contains("trace-xadd-maxlen") so the test verifies
structure rather than a substring.
---
Outside diff comments:
In
`@src/main/java/com/pooli/traffic/service/decision/TrafficDeductOrchestratorService.java`:
- Line 52: startedAt is created with LocalDateTime.now() using the system
default zone while finishedAt uses trafficRedisRuntimePolicy.zoneId(), causing
timezone inconsistency; change startedAt to be created with the same zone as
finishedAt (use trafficRedisRuntimePolicy.zoneId()) so both timestamps are in
the same timezone (e.g., LocalDateTime.now(trafficRedisRuntimePolicy.zoneId())
or switch both to ZonedDateTime.now(trafficRedisRuntimePolicy.zoneId())), and
update any downstream latency/calculation logic in
TrafficDeductOrchestratorService that references startedAt/finishedAt
accordingly.
In
`@src/main/java/com/pooli/traffic/service/outbox/TrafficPolicyVersionedRedisService.java`:
- Around line 86-112: In syncAppPolicySnapshot in
TrafficPolicyVersionedRedisService the keys List is created with new
ArrayList<>() causing potential reallocations; compute an initial capacity
(e.g., 3 + (qosSpeedLimitNextAvailableKeys == null ? 0 :
qosSpeedLimitNextAvailableKeys.size())) and construct the ArrayList with that
capacity before adding appDataKey, appSpeedKey, appWhitelistKey and
qosSpeedLimitNextAvailableKeys to avoid internal array growth.
In
`@src/main/java/com/pooli/traffic/service/policy/TrafficPolicyWriteThroughService.java`:
- Around line 319-373: Replace the LinkedHashSet used for
appIdsToClearReservation in syncAppPolicySnapshotUntracked with a plain HashSet
since iteration order isn't required; update the declaration of
appIdsToClearReservation to use HashSet<Integer> and leave the rest of the logic
that builds qosSpeedLimitNextAvailableKeys unchanged, and remove any unused
LinkedHashSet import from the file.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 5c293554-9f45-48af-8682-bd0815abc452
📒 Files selected for processing (33)
docs/redis-only-release-checklist.mdsrc/main/java/com/pooli/common/config/AppStreamsProperties.javasrc/main/java/com/pooli/traffic/domain/TrafficLuaDeductExecutionResult.javasrc/main/java/com/pooli/traffic/domain/dto/response/TrafficLuaDeductResDto.javasrc/main/java/com/pooli/traffic/domain/enums/TrafficLuaStatus.javasrc/main/java/com/pooli/traffic/service/decision/TrafficDeductLuaExecutor.javasrc/main/java/com/pooli/traffic/service/decision/TrafficDeductOrchestratorService.javasrc/main/java/com/pooli/traffic/service/invoke/TrafficDeductDoneLogService.javasrc/main/java/com/pooli/traffic/service/invoke/TrafficRequestEnqueueService.javasrc/main/java/com/pooli/traffic/service/invoke/TrafficStreamInfraService.javasrc/main/java/com/pooli/traffic/service/outbox/TrafficPolicyVersionedRedisService.javasrc/main/java/com/pooli/traffic/service/policy/TrafficPolicyWriteThroughService.javasrc/main/java/com/pooli/traffic/service/runtime/TrafficLuaScriptInfraService.javasrc/main/java/com/pooli/traffic/service/runtime/TrafficRedisKeyFactory.javasrc/main/resources/application-local.ymlsrc/main/resources/application.yamlsrc/main/resources/lua/traffic/app_policy_single_cas.luasrc/main/resources/lua/traffic/app_policy_snapshot_cas.luasrc/main/resources/lua/traffic/deduct_unified.luasrc/test/java/com/pooli/common/config/AppStreamsPropertiesTest.javasrc/test/java/com/pooli/traffic/acceptance/TrafficDataPolicyAcceptanceTest.javasrc/test/java/com/pooli/traffic/service/TrafficProfileBootTest.javasrc/test/java/com/pooli/traffic/service/decision/TrafficDeductLuaExecutorTest.javasrc/test/java/com/pooli/traffic/service/decision/TrafficDeductOrchestratorServiceTest.javasrc/test/java/com/pooli/traffic/service/invoke/TrafficDeductDoneLogServiceTest.javasrc/test/java/com/pooli/traffic/service/invoke/TrafficRequestEnqueueServiceTest.javasrc/test/java/com/pooli/traffic/service/invoke/TrafficStreamConsumerRunnerTest.javasrc/test/java/com/pooli/traffic/service/invoke/TrafficStreamInfraServiceTest.javasrc/test/java/com/pooli/traffic/service/outbox/TrafficPolicyVersionedRedisServiceTest.javasrc/test/java/com/pooli/traffic/service/policy/TrafficPolicyWriteThroughServiceTest.javasrc/test/java/com/pooli/traffic/service/runtime/TrafficLuaPolicyContractTest.javasrc/test/java/com/pooli/traffic/service/runtime/TrafficLuaScriptInfraServiceTest.javasrc/test/java/com/pooli/traffic/service/runtime/TrafficRedisKeyFactoryTest.java
|
|
||
| ## Redis 서버 버전 확인 | ||
| - 현재 repository의 `docker-compose.yml`은 Redis 서버 이미지를 고정하지 않고, 배포 환경 Redis는 `.env`의 host/port로 주입된다. | ||
| - M5 확인 결과: `.env` 기준 cache Redis와 streams Redis 모두 `INFO server`의 `redis_version`이 `7.4.2`. |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial | ⚡ Quick win
체크리스트에 환경 고정값을 하드코딩하지 않는 편이 안전합니다.
Line 20의 M5 확인 결과 ... redis_version=7.4.2는 시간이 지나면 문서가 쉽게 stale 되어 릴리즈 게이트 신뢰도를 떨어뜨릴 수 있습니다. 체크리스트에는 검증 절차와 기록 위치만 남기고, 실제 버전 값은 릴리즈 기록(배포 노트/PR 코멘트)으로 분리해 주세요.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/redis-only-release-checklist.md` at line 20, 문서의 하드코딩된 Redis 버전("M5 확인
결과 ... redis_version=7.4.2") 때문에 체크리스트가 쉽게 stale 해질 수 있으므로 해당 하드코딩된 버전 문자열을 제거하고
대신 검증 절차와 결과를 어디에 기록할지 명시하세요; 예를 들어 `M5 확인 결과` 항목에는 "`.env` 기준 cache/streams
Redis의 redis_version을 확인" 같은 절차만 남기고, 실제 버전값은 릴리즈 노트나 PR 코멘트(릴리즈 기록 위치)에 기록하도록
안내문을 추가하여 `redis_version` 값은 문서에 남기지 않도록 수정하세요.
Comment on lines
+158
to
+179
| // 1. 먼저 XACK로 consumer group의 PEL에서 record를 제거한다. | ||
| // XACK 실패는 메시지 완료 처리 실패로 보아 호출부로 그대로 전파한다. | ||
| Long acknowledged = executeStreamsOperation(() -> | ||
| streamOps().acknowledge( | ||
| appStreamsProperties.getKeyTrafficRequest(), | ||
| appStreamsProperties.getGroupTraffic(), | ||
| streamKey, | ||
| group, | ||
| recordId | ||
| ) | ||
| ); | ||
| return acknowledged == null ? 0L : acknowledged; | ||
|
|
||
| // 2. RedisTemplate이 null을 반환하면 ACK되지 않은 것과 같은 0건으로 보정한다. | ||
| long acknowledgedCount = acknowledged == null ? 0L : acknowledged; | ||
|
|
||
| if (acknowledgedCount <= 0L) { | ||
| // 3. ACK된 record가 없으면 아직 완료 상태로 확정할 수 없으므로 XDEL을 시도하지 않는다. | ||
| log.warn("traffic_stream_ack_zero streamKey={} group={} recordId={}", streamKey, group, recordId.getValue()); | ||
| return acknowledgedCount; | ||
| } | ||
|
|
||
| // 4. ACK가 성공한 뒤에만 stream entry 삭제를 best-effort로 시도한다. | ||
| deleteAcknowledgedRecord(streamKey, group, recordId); | ||
| return acknowledgedCount; |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial | 🏗️ Heavy lift
ACK 이후 XDEL 순차 처리 시 Redis 왕복 횟수 고려
현재 구현은 XACK 성공 후 XDEL을 별도로 호출하여 메시지 완료당 2번의 Redis 왕복(round-trip)이 발생합니다. 높은 처리량 환경에서는 이로 인한 지연 시간 누적이 전체 처리 성능에 영향을 줄 수 있습니다.
개선 방향 제안:
- Redis Pipelining 활용: 동일 Consumer에서 여러 메시지를 처리할 때 XACK와 XDEL 명령을 파이프라인으로 일괄 전송하면 네트워크 왕복을 크게 줄일 수 있습니다.
- 배치 처리: Consumer가 여러 record를 읽어온 경우, 모든 XACK를 먼저 수행한 후 성공한 recordId 목록을 한 번에 XDEL하는 방식을 고려할 수 있습니다.
다만, PR 목표가 "best-effort 삭제"로 명시되어 있고 XDEL 실패가 완료 흐름을 깨지 않도록 설계된 점을 고려하면, 현재 구현은 의도된 trade-off일 수 있습니다. 처리량이 증가할 경우 모니터링을 통해 병목 여부를 확인하고 필요시 파이프라이닝 적용을 검토하시기 바랍니다.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/main/java/com/pooli/traffic/service/invoke/TrafficStreamInfraService.java`
around lines 158 - 179, The current flow performs XACK then XDEL per record
causing two Redis round-trips; change to batch/pipeline these operations: add a
new method (e.g., deleteAcknowledgedRecordsBatch or
deleteAcknowledgedRecordPipeline) that accepts a collection of recordIds and
uses Redis pipelining or streamOps().executePipelined to send all
acknowledge(...) calls and then a single batched XDEL (or pipelined XDELs) for
the succeeded ids; update the caller that currently invokes
executeStreamsOperation/streamOps().acknowledge and deleteAcknowledgedRecord to
collect recordIds per poll and call the new batch/pipeline method so multiple
records are ACKed and XDELed in a single network round-trip while preserving
best-effort semantics (ignore XDEL failures).
Comment on lines
+171
to
+175
| if (acknowledgedCount <= 0L) { | ||
| // 3. ACK된 record가 없으면 아직 완료 상태로 확정할 수 없으므로 XDEL을 시도하지 않는다. | ||
| log.warn("traffic_stream_ack_zero streamKey={} group={} recordId={}", streamKey, group, recordId.getValue()); | ||
| return acknowledgedCount; | ||
| } |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
ACK 건수 0 처리 로직 확인
acknowledgedCount <= 0인 경우 조기 반환하여 XDEL을 건너뛰는 로직이 추가되었습니다. 이는 안전한 방어 코드이지만, 정상 흐름에서는 발생하지 않아야 하는 상황입니다.
확인 사항:
- 해당 경로가 실제로 발생하는지 운영 로그를 모니터링하시기 바랍니다.
- 만약
traffic_stream_ack_zero로그가 빈번하게 발생한다면:- Consumer group의 상태 불일치
- 중복 ACK 시도
- Redis Streams 버전별 동작 차이
- 등의 근본 원인을 파악해야 합니다.
정상적인 완료 흐름에서는 이 분기에 도달하지 않으므로, 로그 발생 시 알림을 설정하는 것도 고려할 수 있습니다.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/main/java/com/pooli/traffic/service/invoke/TrafficStreamInfraService.java`
around lines 171 - 175, The early-return branch for acknowledgedCount <= 0 (in
TrafficStreamInfraService where acknowledgedCount,
log.warn("traffic_stream_ack_zero", streamKey, group, recordId.getValue()) and
the subsequent skipping of XDEL occur) must be instrumented and surfaced: add a
metric/alert increment when this branch is hit, enrich the log with diagnostic
context (consumer group name, pending count, stream info and relevant IDs), and
ensure any retry/duplicate-ACK logic is audited so operations can investigate;
keep the current return behavior but emit the metric and richer log so frequent
occurrences trigger investigation.
Comment on lines
+352
to
+388
| local finished_at_epoch_millis = redis_epoch_millis() | ||
| if speed_limited_bytes > 0 then | ||
| -- 속도값이 0이면 duration을 계산할 수 없으므로 즉시 timeout으로 기각한다. | ||
| if not effective_bytes_per_sec or effective_bytes_per_sec <= 0 then | ||
| return as_json(0, 0, 0, "SPEED_LIMIT_TIMEOUT", finished_at_epoch_millis) | ||
| end | ||
|
|
||
| -- 예약 키가 이미 있으면 이전 요청의 완료 예정 시각 이후에 이번 요청을 이어 붙인다. | ||
| -- 예약 키가 없거나 과거 시각이면 Redis 현재 시각부터 계산한다. | ||
| local reservation_raw = redis.call("GET", qos_speed_limit_next_available_key) | ||
| local reservation_base = finished_at_epoch_millis | ||
| if reservation_raw ~= false and reservation_raw ~= nil then | ||
| local parsed_reservation = tonumber(reservation_raw) | ||
| if not parsed_reservation or parsed_reservation < 0 then | ||
| return as_json(0, 0, 0, "ERROR", finished_at_epoch_millis) | ||
| end | ||
| reservation_base = math.max(parsed_reservation, finished_at_epoch_millis) | ||
| end | ||
|
|
||
| -- durationMs = ceil(requestBytes * 1000 / bytesPerSec) | ||
| -- finishedAtMs = max(nextAvailableAtMs, nowMs) + durationMs | ||
| local duration_ms = math.ceil(speed_limited_bytes * 1000 / effective_bytes_per_sec) | ||
| local reserved_finished_at_epoch_millis = reservation_base + duration_ms | ||
| -- finishedAtMs가 Redis 현재 시각 기준 5초를 넘으면 전체 요청을 처리하지 않는다. | ||
| -- 이 return은 dedupe 생성, 잔량 차감, usage 증가, 예약 키 갱신보다 앞에 있다. | ||
| if reserved_finished_at_epoch_millis > finished_at_epoch_millis + SPEED_LIMIT_TIMEOUT_MS then | ||
| return as_json(0, 0, 0, "SPEED_LIMIT_TIMEOUT", finished_at_epoch_millis) | ||
| end | ||
|
|
||
| -- 5초 안에 끝날 수 있으면 예약 키를 새 완료 예정 시각으로 갱신한다. | ||
| -- 이후 같은 lineId+appId 요청은 이 시각을 기준으로 다시 이어 붙는다. | ||
| finished_at_epoch_millis = reserved_finished_at_epoch_millis | ||
| redis.call("SET", qos_speed_limit_next_available_key, finished_at_epoch_millis) | ||
| if app_speed_limited then | ||
| final_status = "HIT_APP_SPEED" | ||
| end | ||
| end |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial | 💤 Low value
예약 키에 TTL이 설정되지 않음.
qos_speed_limit_next_available_key가 SET 명령으로 저장되지만 TTL이 없습니다. 정책 동기화 시 삭제되긴 하지만, 정책이 활성 상태로 유지되면서 해당 회선+앱 조합이 더 이상 사용되지 않는 경우 키가 무기한 남을 수 있습니다.
다만, 새 요청 시 max(parsed_reservation, finished_at_epoch_millis) 로직으로 과거 값은 무시되므로 기능상 문제는 없습니다. 메모리 관리 관점에서 적절한 TTL(예: 1일) 설정을 권장드립니다.
♻️ TTL 설정 제안
- redis.call("SET", qos_speed_limit_next_available_key, finished_at_epoch_millis)
+ -- 예약 키 TTL: 1일 (86400초). 활성 요청이 있으면 갱신되므로 안전함.
+ redis.call("SET", qos_speed_limit_next_available_key, finished_at_epoch_millis, "EX", 86400)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/main/resources/lua/traffic/deduct_unified.lua` around lines 352 - 388,
The reservation key qos_speed_limit_next_available_key is written with
redis.call("SET", qos_speed_limit_next_available_key, finished_at_epoch_millis)
but no TTL is set, which can leave stale keys indefinitely; change the write to
set an expiration (e.g., 86400 seconds) by using Redis TTL-aware command (SET
with EX or SETEX) when updating qos_speed_limit_next_available_key (or introduce
a constant like SPEED_LIMIT_TTL_SECONDS) so the key expires if the line+app is
unused; keep the same value logic (finished_at_epoch_millis) and ensure the new
TTL is applied whenever reservation is updated in this code path.
Comment on lines
+69
to
+76
| @Test | ||
| @DisplayName("traffic request stream max length가 0 이하이면 예외를 던진다") | ||
| void throwsWhenTrafficRequestMaxLengthIsNotPositive() { | ||
| AppStreamsProperties properties = new AppStreamsProperties(); | ||
| properties.setTrafficRequestMaxLength(0L); | ||
|
|
||
| assertThrows(TrafficStreamBootstrapException.class, properties::requireTrafficRequestMaxLength); | ||
| } |
There was a problem hiding this comment.
0 이하 계약이면 음수 케이스도 함께 고정해 주세요.
현재는 0만 검증해서 음수 입력 회귀를 놓칠 수 있습니다. 0과 -1을 파라미터화하면 경계 계약이 더 명확해집니다.
✅ 제안 diff
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
@@
- `@Test`
+ `@ParameterizedTest`
+ `@ValueSource`(longs = {0L, -1L})
`@DisplayName`("traffic request stream max length가 0 이하이면 예외를 던진다")
- void throwsWhenTrafficRequestMaxLengthIsNotPositive() {
+ void throwsWhenTrafficRequestMaxLengthIsNotPositive(long invalidValue) {
AppStreamsProperties properties = new AppStreamsProperties();
- properties.setTrafficRequestMaxLength(0L);
+ properties.setTrafficRequestMaxLength(invalidValue);
assertThrows(TrafficStreamBootstrapException.class, properties::requireTrafficRequestMaxLength);
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/test/java/com/pooli/common/config/AppStreamsPropertiesTest.java` around
lines 69 - 76, Update the test in AppStreamsPropertiesTest to cover both zero
and negative boundary values: call
AppStreamsProperties#setTrafficRequestMaxLength with 0 and with a negative value
(e.g., -1) and assert that AppStreamsProperties#requireTrafficRequestMaxLength
throws TrafficStreamBootstrapException for each; convert the current single-case
test into a parameterized test (or add a second test) that iterates the invalid
inputs so both 0 and negative inputs are verified.
Comment on lines
+198
to
+200
| long requestBytes = 1_300_000L; | ||
| int speedBytesPerSecond = 625_000; | ||
| long expectedDurationMs = 2_080L; |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial | ⚡ Quick win
예상 처리시간 상수는 입력값 기반 계산으로 바꿔 주세요.
expectedDurationMs를 하드코딩하면 requestBytes/speedBytesPerSecond 변경 시 테스트가 조용히 불일치할 수 있습니다. 계산식으로 두면 시나리오 의도와 유지보수성이 같이 올라갑니다.
♻️ 제안 diff
- long expectedDurationMs = 2_080L;
+ long expectedDurationMs = (long) Math.ceil((double) requestBytes * 1000 / speedBytesPerSecond);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| long requestBytes = 1_300_000L; | |
| int speedBytesPerSecond = 625_000; | |
| long expectedDurationMs = 2_080L; | |
| long requestBytes = 1_300_000L; | |
| int speedBytesPerSecond = 625_000; | |
| long expectedDurationMs = (long) Math.ceil((double) requestBytes * 1000 / speedBytesPerSecond); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/test/java/com/pooli/traffic/acceptance/TrafficDataPolicyAcceptanceTest.java`
around lines 198 - 200, Replace the hardcoded expectedDurationMs with a
calculated value derived from requestBytes and speedBytesPerSecond: compute the
transfer time in milliseconds using requestBytes and speedBytesPerSecond (e.g.,
ceil((double)requestBytes / speedBytesPerSecond * 1000) or integer math like
(requestBytes * 1000 + speedBytesPerSecond - 1) / speedBytesPerSecond) and
assign that result to expectedDurationMs so the test stays correct when
requestBytes or speedBytesPerSecond change; update the code that defines
expectedDurationMs accordingly (referencing the variables requestBytes and
speedBytesPerSecond).
| assertEquals(500_000L, options.getMaxlen()); | ||
| assertTrue(options.isApproximateTrimming()); | ||
| assertEquals("traffic:deduct:request", record.getStream()); | ||
| assertTrue(String.valueOf(((Map<?, ?>) record.getValue()).get("payload")).contains("trace-xadd-maxlen")); |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial | ⚡ Quick win
payload 검증은 문자열 포함 검사 대신 구조 검증으로 고정해 주세요.
현재 contains("trace-xadd-maxlen")는 우연히 문자열이 포함돼도 통과할 수 있습니다. JSON 파싱 후 traceId 필드를 직접 검증하면 테스트 신뢰도가 높아집니다.
♻️ 제안 diff
- assertTrue(String.valueOf(((Map<?, ?>) record.getValue()).get("payload")).contains("trace-xadd-maxlen"));
+ String payloadJson = String.valueOf(((Map<?, ?>) record.getValue()).get("payload"));
+ assertEquals("trace-xadd-maxlen", new ObjectMapper().readTree(payloadJson).get("traceId").asText());🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/test/java/com/pooli/traffic/service/invoke/TrafficRequestEnqueueServiceTest.java`
at line 86, The current assertion uses a brittle substring check on ((Map<?,?>)
record.getValue()).get("payload"); instead parse the payload JSON and validate
the traceId field directly: extract payload from record.getValue(), parse it
(e.g., with your test ObjectMapper) into a Map or DTO, then assert the traceId
field exists and equals or matches the expected value/pattern. Update the
assertion in TrafficRequestEnqueueServiceTest to use the parsed payload's
"traceId" instead of contains("trace-xadd-maxlen") so the test verifies
structure rather than a substring.
Owner
Author
|
@CodeRabbit review |
✅ Actions performedReview triggered.
|
This was referenced May 27, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
개요
본 PR은 데이터 차감 로직의 전면적인 Redis-Only 전환 및 속도 제한(QoS, 앱 속도 제한) 메커니즘 통합 개편을 포함합니다. 주요 변경 목적은 복잡했던 DB Fallback 및 Refill 단계를 완전히 폐기하고, Redis 기반의 단일 흐름으로 정합성과 속도를 높이기 위함입니다.
주요 변경 사항:
SPEED_LIMIT_TIMEOUT기각 상태를 반환하도록 변경하여 지연 응답 문제를 해결했습니다.PENDING기록을 하나의 DB 트랜잭션으로 묶고, 처리량은 개인/공유/QoS로 명확히 분리하여 멱등성 및 재시도 정합성을 보장했습니다.MAXLEN ~ 500000) 제한을 활성화하고, ACK 성공 직후 해당 Stream Entry를 즉각 삭제(XDEL) 시도합니다. 더불어 Redis 상태(cache, streams)를 분리해 PING 기반의 능동 가용성 경보(Alertmanager) 체계도 구축했습니다.관련 BackLog
Resolves: (데이터 차감 Redis-Only 전환, QoS 속도 제한 단일화, Redis Stream 최적화, Outbox 트랜잭션 강화)
PR 유형
PR Checklist
📊 통합 차감 및 데이터 정합성 시퀀스 다이어그램
새롭게 개편된 단일 Lua 차감 및 Outbox DB 트랜잭션 동기화 로직의 흐름은 다음과 같습니다.
%%{init: {"theme": "default", "themeVariables": {"background": "#ffffff"}}}%% sequenceDiagram autonumber participant Worker as TrafficStreamConsumerRunner participant Lua as 단일 차감 Lua (Redis) participant Persistence as TrafficDeductCompletionPersistenceService participant DB as MySQL DB Worker->>Lua: 단일 차감 시도 (개인풀 -> 공유풀 -> QoS) Note over Lua: QoS와 앱 속도 제한 동시 적용 시<br/>더 작은 속도를 기준으로 예약 키(TIME) 계산 alt 5초 내 처리 불가능 (Timeout) Lua-->>Worker: SPEED_LIMIT_TIMEOUT 기각 반환 else 5초 내 처리 가능 Lua-->>Worker: 차감 완료 결과 및 finishedAtEpochMillis 반환 end Worker->>Persistence: Done Log 및 Outbox 생성 요청 rect rgb(230, 240, 255) Note right of Persistence: DB 트랜잭션 시작 Persistence->>DB: Done Log Insert (Redis 완료 시각 보존) Persistence->>DB: Dedupe 제거용 Outbox PENDING 생성 Note left of DB: DB 트랜잭션 커밋 완료 end Persistence-->>Worker: 생성된 Outbox ID 반환 Worker->>Lua: Redis Dedupe Key 즉시 삭제 시도 alt 즉시 삭제 성공 Worker->>DB: Outbox 상태를 SUCCESS 로 업데이트 else 즉시 삭제 실패 Worker->>DB: Outbox 상태를 FAIL 로 업데이트 (이후 Scheduler가 재시도) end Worker->>Lua: 메시지 XACK 및 XDEL 요청 Note over Lua: XACK 성공 시 즉시 해당 Entry 삭제 시도<br/>(스트림 보관 길이 제한 MAXLEN ~ 500000 유지 방어)