[moonwhistle]

summary

체크포인트를 consumer 복구의 기준으로, 배치 검증을 최종 봉 정합성의 기준으로 분리합니다.

changes

• 봉·집계 상태·처리 위치를 동일한 DB 트랜잭션으로 저장하고, 커밋 이후 ACK하도록 처리 경계를 통일합니다.
• 재기동 시 체크포인트 기반 원본 재생으로 집계 상태를 복구하고, 저장·검증 근거가 있는 메시지만 PEL에서 정리합니다.
• 실패 기록을 먼저 영속화하고 DLQ 전달과 원본 ACK를 Redis Lua로 묶습니다. 제한된 backoff 재시도 이후에는 PEL을 유지하며, DLQ 전달 상태와 배치 보정 상태를 별도로 관리합니다.
• 배치 검증 결과와 consumer 쓰기를 공통 저장 경계로 보호하여, 늦은 처리가 보정된 봉을 덮어쓰지 못하도록 합니다.
• 원본 보존 범위를 체크포인트·미처리 PEL에 연동하고, collector에는 선택 가능한 WAL 기반 전달 경로를 추가합니다.

note

• 단일 활성 consumer를 전제로 하며, 기존 consumer group의 최초 전환에는 명시적인 복구 기준 설정이 필요합니다.
• DLQ는 실패 조사·보정 추적용입니다. 별도 tick 재집계 워커나 범용 outbox는 두지 않습니다.
• Lua 실행은 실패 시 롤백을 보장하지 않으며, WAL은 프로세스 장애 복구용입니다. 저장소 자체 유실까지 포함한 exactly-once 보장은 범위 밖입니다.

[qodo-code-review]

ⓘ Qodo reviews are paused because your trial has ended. Ask your workspace admin to add credits to resume reviews. Manage billing

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated
coin-flow	Ready	Preview	Sep 21, 2026 7:47am UTC

[coderabbitai]

Important

Review skipped

Too many files!

This PR contains 119 files, which is 19 over the limit of 100.

To get a review, reduce the PR to 100 files or fewer by splitting it into smaller PRs or changing its base branch.

Upgrade to a paid plan to raise the limit.

This review couldn't start because sufficient usage credits or metered capacity aren't available. Add credits or update usage-based reviews in the billing tab, then retry.

⚙️ Run configuration

Configuration used: Repository: moonwhistle/CoinFlow/.coderabbit.yaml

Review profile: CHILL

Plan: Advanced

Run ID: b0a77f11-8971-4751-a90b-e4c4cf1f08d2

📥 Commits

Reviewing files that changed from the base of the PR and between 3c2157c and 4bddbcb.

⛔ Files ignored due to path filters (2)

• benchmark-results/collector-wal-20260910-interleaved/results.csv is excluded by !**/*.csv
• benchmark-results/collector-wal-20260910-metrics-supplement/results.csv is excluded by !**/*.csv

📒 Files selected for processing (119)

• .gitignore
• backend/coinflow-collector-app/src/main/java/com/coinflow/handler/binance/BinanceTradeMessageHandler.java
• backend/coinflow-collector-app/src/main/java/com/coinflow/handler/binance/constant/BinanceTradeMessageFields.java
• backend/coinflow-collector-app/src/main/resources/application-collector.yml
• backend/coinflow-collector-app/src/test/java/com/coinflow/handler/binance/BinanceTradeMessageHandlerTest.java
• backend/coinflow-common/src/main/java/com/coinflow/tick/serialization/TickRawBinaryCodec.java
• backend/coinflow-common/src/test/java/com/coinflow/tick/serialization/TickRawBinaryCodecTest.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/infrastructure/persistence/DbPersistService.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/BatchAckWorker.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/TickProcessService.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/config/RedisConsumerConfig.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/config/RedisConsumerGroupManager.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/config/properties/TickConsumerProperties.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/handler/TickRawMessageHandler.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/monitoring/StreamLagMonitorWorker.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/recovery/service/CandleProjectionPublisher.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/recovery/service/ConsumerCheckpointService.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/recovery/service/FailedRecordService.java
• backend/coinflow-consumer-app/src/main/java/com/coinflow/recovery/service/StreamRecoveryService.java
• backend/coinflow-consumer-app/src/main/resources/application-consumer.yml
• backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/BatchAckWorkerTest.java
• backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/TickProcessServiceTest.java
• backend/coinflow-consumer-app/src/test/java/com/coinflow/config/RedisConsumerGroupManagerTest.java
• backend/coinflow-consumer-app/src/test/java/com/coinflow/handler/TickRawMessageHandlerTest.java
• backend/coinflow-consumer-app/src/test/java/com/coinflow/recovery/service/ConsumerStartupIntegrationTest.java
• backend/coinflow-consumer-app/src/test/java/com/coinflow/recovery/service/FailedRecordServiceTest.java
• backend/coinflow-consumer-app/src/test/java/com/coinflow/recovery/service/RecoveryIntegrationTest.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/aggregation/domain/KlineState.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/aggregation/domain/MutableKlineSnapshot.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/aggregation/service/KlineAggregatorService.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/domain/ConsumerCheckpoint.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/domain/FailedRecord.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/domain/StreamRecordIds.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/domain/VerifiedCandle.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/repository/ConsumerCheckpointRepository.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/repository/FailedRecordRepository.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/repository/RecoveryStreamRepository.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/repository/VerifiedCandleRepository.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/service/RecoveryCandleStore.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/service/RecoveryLedger.java
• backend/coinflow-core/src/main/java/com/coinflow/domain/recovery/service/RecoveryMaintenanceService.java
• backend/coinflow-core/src/test/java/com/coinflow/domain/aggregation/service/AggregationCheckpointTest.java
• backend/coinflow-infra-redis/build.gradle
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/DeliveryStatusProvider.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/config/CollectorDeliveryProperties.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/config/TickPublisherConfig.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/stream/QueuedPipelinedTickPublisher.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/stream/RedisStreamTickPublisher.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/wal/LocalTickWal.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/wal/WalCheckpoint.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/wal/WalPipelinedTickPublisher.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/publish/wal/WalRecord.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/recovery/redis/RecoveryMaintenanceWorker.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/recovery/redis/RedisDeadLetterStore.java
• backend/coinflow-infra-redis/src/main/java/com/coinflow/recovery/redis/RedisRecoveryStreamRepository.java
• backend/coinflow-infra-redis/src/test/java/com/coinflow/publish/stream/QueuedPipelinedTickPublisherTest.java
• backend/coinflow-infra-redis/src/test/java/com/coinflow/publish/wal/LocalTickWalTest.java
• backend/coinflow-infra-redis/src/test/java/com/coinflow/publish/wal/WalPipelinedTickPublisherTest.java
• backend/coinflow-infra-redis/src/test/java/com/coinflow/recovery/redis/RedisDeadLetterStoreIntegrationTest.java
• backend/coinflow-load-test/build.gradle
• backend/coinflow-load-test/src/main/java/com/coinflow/loadtest/CollectorDeliveryBenchmarkMain.java
• backend/coinflow-load-test/src/main/java/com/coinflow/loadtest/CollectorWalRecoveryProbeMain.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/config/ReconciliationJobConfig.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/processor/BinanceKlineProcessor.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/processor/OhlcRollupProcessor.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/scheduler/PendingRepairScheduler.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/scheduler/ReconciliationScheduler.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/writer/BinanceKlineWriter.java
• backend/coinflow-replay-app/src/main/java/com/coinflow/replay/batch/writer/OhlcRollupWriter.java
• backend/coinflow-replay-app/src/main/resources/application-replay.yml
• backend/coinflow-replay-app/src/test/java/com/coinflow/replay/batch/ReconciliationJobIntegrationTest.java
• backend/coinflow-replay-app/src/test/java/com/coinflow/replay/batch/processor/BinanceKlineProcessorTest.java
• backend/coinflow-replay-app/src/test/java/com/coinflow/replay/batch/processor/OhlcRollupProcessorTest.java
• backend/coinflow-replay-app/src/test/java/com/coinflow/replay/batch/scheduler/PendingRepairSchedulerTest.java
• backend/testscript/run_collector_delivery_benchmark.ps1
• benchmark-results/collector-wal-20260910-interleaved/README.md
• benchmark-results/collector-wal-20260910-interleaved/direct-tps1000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps1000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps1000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps10000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps10000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps10000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps5000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps5000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tps5000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tpsmax-r1.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tpsmax-r2.json
• benchmark-results/collector-wal-20260910-interleaved/direct-tpsmax-r3.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps1000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps1000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps1000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps10000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps10000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps10000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps5000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps5000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tps5000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tpsmax-r1.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tpsmax-r2.json
• benchmark-results/collector-wal-20260910-interleaved/pipeline-tpsmax-r3.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps1000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps1000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps1000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps10000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps10000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps10000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps5000-r1.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps5000-r2.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tps5000-r3.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tpsmax-r1.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tpsmax-r2.json
• benchmark-results/collector-wal-20260910-interleaved/wal_pipeline-tpsmax-r3.json
• benchmark-results/collector-wal-20260910-metrics-supplement/README.md
• benchmark-results/collector-wal-20260910-metrics-supplement/direct-tps10000-r1.json
• benchmark-results/collector-wal-20260910-metrics-supplement/pipeline-tps10000-r1.json
• benchmark-results/collector-wal-20260910-metrics-supplement/wal_pipeline-tps10000-r1.json
• docs/consumer-recovery.md
• infra/docker/docker-compose-prod.yml
• infra/sql/consumer-recovery.sql

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}