diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/ApiApplication.java b/backend/coinflow-api-app/src/main/java/com/coinflow/ApiApplication.java index 38ea9e77..8d1d08ab 100644 --- a/backend/coinflow-api-app/src/main/java/com/coinflow/ApiApplication.java +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/ApiApplication.java @@ -2,8 +2,10 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication +@EnableScheduling public class ApiApplication { public static void main(String[] args) { diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/cache/hot/OhlcHotWindow.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/cache/hot/OhlcHotWindow.java new file mode 100644 index 00000000..ae42250d --- /dev/null +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/cache/hot/OhlcHotWindow.java @@ -0,0 +1,32 @@ +package com.coinflow.chart.cache.hot; + +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.event.kline.KlineEvent; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +public record OhlcHotWindow( + List finalizedCandles, + KlineEvent liveCandle, + Instant synchronizedAt, + long eventVersion +) { + + public OhlcHotWindow { + finalizedCandles = List.copyOf(finalizedCandles); + } + + public Optional liveCandleOptional() { + return Optional.ofNullable(liveCandle); + } + + public List findFinalizedRange(long toExclusive, int limit) { + List matches = finalizedCandles.stream() + .filter(candle -> candle.epochSeconds() < toExclusive) + .toList(); + int fromIndex = Math.max(0, matches.size() - limit); + return new ArrayList<>(matches.subList(fromIndex, matches.size())); + } +} diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/cache/hot/OhlcHotWindowStore.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/cache/hot/OhlcHotWindowStore.java new file mode 100644 index 00000000..fc2b284f --- /dev/null +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/cache/hot/OhlcHotWindowStore.java @@ -0,0 +1,136 @@ +package com.coinflow.chart.cache.hot; + +import com.coinflow.domain.ohlc.constant.OhlcWindowPolicy; +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.event.kline.KlineEvent; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.binder.cache.CaffeineCacheMetrics; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.springframework.stereotype.Component; + +@Component +public class OhlcHotWindowStore { + + private static final int MAX_CACHE_KEYS = 500; + + private final Cache cache; + + public OhlcHotWindowStore(MeterRegistry meterRegistry) { + cache = Caffeine.newBuilder() + .maximumSize(MAX_CACHE_KEYS) + .expireAfterAccess(10, TimeUnit.MINUTES) + .recordStats() + .build(); + CaffeineCacheMetrics.monitor(meterRegistry, cache, "ohlc_hot_window_cache"); + } + + public Optional get(String symbol, String interval) { + return Optional.ofNullable(cache.getIfPresent(key(symbol, interval))); + } + + public void replace( + String symbol, + String interval, + List finalizedCandles, + Optional liveCandle, + Instant synchronizedAt + ) { + KlineEvent openLiveCandle = liveCandle.filter(event -> !event.closed()).orElse(null); + cache.put( + key(symbol, interval), + new OhlcHotWindow(normalize(finalizedCandles), openLiveCandle, synchronizedAt, 0) + ); + } + + public long eventVersion(String symbol, String interval) { + OhlcHotWindow current = cache.getIfPresent(key(symbol, interval)); + return current == null ? 0 : current.eventVersion(); + } + + public boolean replaceIfVersion( + String symbol, + String interval, + List finalizedCandles, + Optional liveCandle, + Instant synchronizedAt, + long expectedVersion + ) { + String key = key(symbol, interval); + KlineEvent openLiveCandle = liveCandle.filter(event -> !event.closed()).orElse(null); + AtomicBoolean replaced = new AtomicBoolean(); + cache.asMap().compute(key, (ignored, current) -> { + long currentVersion = current == null ? 0 : current.eventVersion(); + if (currentVersion != expectedVersion) { + return current; + } + replaced.set(true); + return new OhlcHotWindow( + normalize(finalizedCandles), openLiveCandle, synchronizedAt, currentVersion); + }); + return replaced.get(); + } + + public void applyEvent(KlineEvent event) { + String key = key(event.symbol(), event.interval()); + cache.asMap().compute(key, (ignored, current) -> { + OhlcHotWindow base = current != null + ? current + : new OhlcHotWindow(List.of(), null, Instant.EPOCH, 0); + + if (!event.closed()) { + return new OhlcHotWindow( + base.finalizedCandles(), event, base.synchronizedAt(), base.eventVersion() + 1); + } + + List updated = new ArrayList<>(base.finalizedCandles()); + updated.removeIf(candle -> candle.epochSeconds() == event.startTime()); + updated.add(toSnapshot(event)); + KlineEvent live = base.liveCandleOptional() + .filter(candidate -> candidate.startTime() != event.startTime()) + .orElse(null); + return new OhlcHotWindow( + normalize(updated), live, base.synchronizedAt(), base.eventVersion() + 1); + }); + } + + private List normalize(List candles) { + Map byTimestamp = new LinkedHashMap<>(); + candles.stream() + .sorted(Comparator.comparingLong(OhlcCandleSnapshot::epochSeconds)) + .forEach(candle -> byTimestamp.put(candle.epochSeconds(), candle)); + + List normalized = new ArrayList<>(byTimestamp.values()); + int fromIndex = Math.max(0, normalized.size() - OhlcWindowPolicy.MAX_SIZE); + return List.copyOf(normalized.subList(fromIndex, normalized.size())); + } + + private OhlcCandleSnapshot toSnapshot(KlineEvent event) { + LocalDateTime bucketTime = LocalDateTime.ofEpochSecond( + event.startTime(), 0, ZoneOffset.UTC); + return new OhlcCandleSnapshot( + bucketTime, + event.startTime(), + event.open(), + event.high(), + event.low(), + event.close(), + event.volume() + ); + } + + private String key(String symbol, String interval) { + return symbol.toLowerCase() + ":" + interval; + } +} diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/constant/ChartCacheConstants.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/constant/ChartCacheConstants.java index 66a22400..9617aabb 100644 --- a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/constant/ChartCacheConstants.java +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/constant/ChartCacheConstants.java @@ -10,16 +10,6 @@ @NoArgsConstructor(access = AccessLevel.PRIVATE) public final class ChartCacheConstants { - /** - * The sliding window size for the global Redis ZSET cache. - */ - public static final int MAX_HOT_WINDOW_SIZE = 1000; - - /** - * Redis key prefix for OHLC candle windows. - */ - public static final String REDIS_WINDOW_KEY_PREFIX = "klines:window:"; - /** * Lock key prefix for preventing Thundering Herd during cache backfill. */ @@ -30,6 +20,11 @@ public final class ChartCacheConstants { */ public static final long LOCK_TIMEOUT_SECONDS = 3; + /** + * Maximum age of a local hot-window snapshot before request-time Redis fallback. + */ + public static final long HOT_WINDOW_STALE_AFTER_MILLIS = 3000; + /** * Property name to enable/disable chart cache warm-up at startup. */ diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/repository/RedisOhlcWindowRepository.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/repository/RedisOhlcWindowRepository.java deleted file mode 100644 index be7dd911..00000000 --- a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/repository/RedisOhlcWindowRepository.java +++ /dev/null @@ -1,36 +0,0 @@ -package com.coinflow.chart.repository; - -import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; -import java.util.List; - -/** - * Interface for managing the Redis Sorted Set (ZSET) based OHLC window. - * Stores the last N closed candles for a given symbol and interval. - */ -public interface RedisOhlcWindowRepository { - - /** - * Saves a candle snapshot to the Redis ZSET. - * Uses bucket startTime (epoch seconds) as the score to ensure O(log N) updates. - */ - void save(String symbol, String interval, OhlcCandleSnapshot snapshot); - - /** - * Batch saves multiple candle snapshots to the Redis ZSET. - * Useful for initial DB backfilling (Gap-fill). - */ - void saveAll(String symbol, String interval, List snapshots); - - /** - * Retrieves a range of candle snapshots from the Redis ZSET. - * @param to The end timestamp (exclusive) for the query. - * @param limit The number of candles to retrieve. - */ - List findRange(String symbol, String interval, long to, int limit); - - /** - * Trims the Redis ZSET to maintain only the last N items. - * Usually called after saving a new candle. - */ - void trim(String symbol, String interval, int limit); -} diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/repository/RedisOhlcWindowRepositoryImpl.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/repository/RedisOhlcWindowRepositoryImpl.java deleted file mode 100644 index 3825d87d..00000000 --- a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/repository/RedisOhlcWindowRepositoryImpl.java +++ /dev/null @@ -1,122 +0,0 @@ -package com.coinflow.chart.repository; - -import com.coinflow.chart.constant.ChartCacheConstants; -import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.ZSetOperations; -import org.springframework.stereotype.Repository; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Objects; -import java.util.Set; - -/** - * Redis Sorted Set (ZSET) implementation of OhlcWindowRepository. - * Stores candle snapshots indexed by epoch seconds for O(log N) retrieval and updates. - */ -@Slf4j -@Repository -@RequiredArgsConstructor -public class RedisOhlcWindowRepositoryImpl implements RedisOhlcWindowRepository { - - private final RedisTemplate redisTemplate; - private final ObjectMapper objectMapper; - - @SuppressWarnings("ConstantConditions") - @Override - public void save(String symbol, String interval, OhlcCandleSnapshot snapshot) { - String key = buildKey(symbol, interval); - String json = serialize(snapshot); - if (json != null) { - ZSetOperations zSetOps = redisTemplate.opsForZSet(); - zSetOps.removeRangeByScore(key, (double) snapshot.epochSeconds(), (double) snapshot.epochSeconds()); - zSetOps.add(key, json, (double) snapshot.epochSeconds()); - log.trace("[REDIS-WINDOW] Saved candle for {} {}: {}", symbol, interval, snapshot.bucketTime()); - } - } - - @Override - public void saveAll(String symbol, String interval, List snapshots) { - if (snapshots == null || snapshots.isEmpty()) return; - - String key = buildKey(symbol, interval); - ZSetOperations zSetOps = redisTemplate.opsForZSet(); - - // Batch remove existing scores to prevent duplicates - long minScore = snapshots.get(0).epochSeconds(); - long maxScore = snapshots.get(snapshots.size() - 1).epochSeconds(); - zSetOps.removeRangeByScore(key, Math.min(minScore, maxScore), Math.max(minScore, maxScore)); - - snapshots.forEach(s -> { - String json = serialize(s); - if (json != null) { - zSetOps.add(key, json, (double) s.epochSeconds()); - } - }); - log.debug("[REDIS-WINDOW] Batch saved {} candles for {} {}", snapshots.size(), symbol, interval); - } - - @SuppressWarnings("ConstantConditions") - @Override - public List findRange(String symbol, String interval, long to, int limit) { - String key = buildKey(symbol, interval); - - // ZREVRANGEBYSCORE: Get 'limit' items from the 'to' timestamp (exclusive) downwards. - Set jsonSet = redisTemplate.opsForZSet() - .reverseRangeByScore(key, -1, (double) to - 1, 0, limit); - - if (jsonSet == null || jsonSet.isEmpty()) { - return Collections.emptyList(); - } - - // Deserialize and reverse back to maintain ascending time order for the chart - List results = jsonSet.stream() - .map(this::deserialize) - .filter(Objects::nonNull) - .toList(); - - // Reverse manually since toList() returns an unmodifiable list in newer Java - List modifiableResults = new ArrayList<>(results); - Collections.reverse(modifiableResults); - return modifiableResults; - } - - @Override - public void trim(String symbol, String interval, int limit) { - String key = buildKey(symbol, interval); - // Maintain only the last N items (e.g., 1000) - Long size = redisTemplate.opsForZSet().size(key); - if (size != null && size > limit) { - redisTemplate.opsForZSet().removeRange(key, 0, size - limit - 1); - log.trace("[REDIS-WINDOW] Trimmed window for {} {}: kept last {}", symbol, interval, limit); - } - } - - private String buildKey(String symbol, String interval) { - return ChartCacheConstants.REDIS_WINDOW_KEY_PREFIX + symbol + ":" + interval; - } - - private String serialize(OhlcCandleSnapshot snapshot) { - try { - return objectMapper.writeValueAsString(snapshot); - } catch (JsonProcessingException e) { - log.error("Failed to serialize OhlcCandleSnapshot: {}", snapshot, e); - return null; - } - } - - private OhlcCandleSnapshot deserialize(String json) { - try { - return objectMapper.readValue(json, OhlcCandleSnapshot.class); - } catch (JsonProcessingException e) { - log.error("Failed to deserialize OhlcCandleSnapshot JSON: {}", json, e); - return null; - } - } -} diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/OhlcChartService.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/OhlcChartService.java index 4636c626..b6910682 100644 --- a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/OhlcChartService.java +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/OhlcChartService.java @@ -1,11 +1,14 @@ package com.coinflow.chart.service; +import com.coinflow.chart.cache.hot.OhlcHotWindow; +import com.coinflow.chart.cache.hot.OhlcHotWindowStore; import com.coinflow.chart.constant.ChartCacheConstants; -import com.coinflow.chart.repository.RedisOhlcWindowRepository; import com.coinflow.domain.ohlc.cache.OhlcChartStore; import com.coinflow.domain.ohlc.constant.OhlcInterval; +import com.coinflow.domain.ohlc.constant.OhlcWindowPolicy; import com.coinflow.domain.ohlc.domain.AbstractOhlc; import com.coinflow.domain.ohlc.repository.LiveKlineRepository; +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; import com.coinflow.domain.ohlc.service.Ohlc1mService; import com.coinflow.domain.ohlc.service.Ohlc30mService; import com.coinflow.domain.ohlc.service.Ohlc5mService; @@ -15,6 +18,7 @@ import com.coinflow.event.kline.KlineEvent; import com.coinflow.util.TimeBucket; import java.time.Clock; +import java.time.Duration; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; @@ -29,16 +33,16 @@ import org.springframework.stereotype.Service; /** - * High-performance chart service using a tiered caching strategy. - * L1 (Caffeine - History) -> L1.5 (Redis ZSET - Hot Window) -> DB (Cold Storage). + * Chart service using local hot/history caches, Redis global windows, and DB fallback. */ @Slf4j @Service @RequiredArgsConstructor public class OhlcChartService { - private final OhlcChartStore chartStore; // Caffeine L2 (History) - private final RedisOhlcWindowRepository ohlcWindowRepository; // Global L1.5 Window + private final OhlcChartStore chartStore; + private final OhlcWindowRepository ohlcWindowRepository; + private final OhlcHotWindowStore hotWindowStore; private final Clock clock; private final Ohlc1mService ohlc1mService; private final Ohlc5mService ohlc5mService; @@ -50,7 +54,7 @@ public class OhlcChartService { /** * Retrieves OHLC candles with high performance. - * Uses Redis-first strategy for the last 1000 candles with DB backfill. + * Uses the local hot window for the last 1000 candles with Redis and DB fallback. * Prevents Thundering Herd via local Mutex and Double-Checked Locking. */ public List show(Long symbolId, OhlcInterval interval, int candles, LocalDateTime to) { @@ -65,15 +69,26 @@ public List show(Long symbolId, OhlcInterval interval, int c // Check if the request is within the 'Hot Window' range (last 1000) long currentEpoch = TimeBucket.to1m(clock.instant()).toEpochSecond(ZoneOffset.UTC); - boolean isHotPath = (currentEpoch - endEpoch) <= interval.duration().getSeconds() * ChartCacheConstants.MAX_HOT_WINDOW_SIZE; + boolean isHotPath = (currentEpoch - endEpoch) + <= interval.duration().getSeconds() * OhlcWindowPolicy.MAX_SIZE; List finalizedCandles; + Optional liveCandle = Optional.empty(); if (isHotPath) { - // Hot Path: Try Redis ZSET - finalizedCandles = ohlcWindowRepository.findRange(symbol.getSymbol(), interval.name(), endEpoch, candles); + Optional localWindow = hotWindowStore + .get(symbol.getSymbol(), interval.name()) + .filter(this::isFresh); + + if (localWindow.isPresent()) { + finalizedCandles = localWindow.get().findFinalizedRange(endEpoch, candles); + liveCandle = localWindow.get().liveCandleOptional(); + } else { + OhlcHotWindow redisWindow = loadRedisHotWindow(symbol, interval, endEpoch); + finalizedCandles = redisWindow.findFinalizedRange(endEpoch, candles); + liveCandle = redisWindow.liveCandleOptional(); + } - // Thundering Herd Prevention: If Redis is empty, acquire lock and backfill if (finalizedCandles.size() < candles) { String lockKey = ChartCacheConstants.LOCK_KEY_PREFIX + symbol.getSymbol() + ":" + interval.name(); ReentrantLock lock = windowLocks.computeIfAbsent(lockKey, k -> new ReentrantLock()); @@ -83,17 +98,30 @@ public List show(Long symbolId, OhlcInterval interval, int c if (lock.tryLock(ChartCacheConstants.LOCK_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { try { // Double-Check: Has another thread already filled the cache? - finalizedCandles = ohlcWindowRepository.findRange(symbol.getSymbol(), interval.name(), endEpoch, candles); + OhlcHotWindow redisWindow = loadRedisHotWindow(symbol, interval, endEpoch); + finalizedCandles = redisWindow.findFinalizedRange(endEpoch, candles); + liveCandle = redisWindow.liveCandleOptional(); if (finalizedCandles.size() < candles) { - log.debug("[CHART-SERVICE] Redis miss/gap for {} {}. Backfilling...", symbol.getSymbol(), interval); - finalizedCandles = backfillAndLoad(symbol, interval, candles, endExclusive); + log.debug( + "[CHART-SERVICE] Redis miss/gap for {} {}. Backfilling...", + symbol.getSymbol(), interval); + finalizedCandles = backfillAndLoad( + symbol, interval, candles, endExclusive); + OhlcHotWindow refreshedWindow = loadRedisHotWindow(symbol, interval, endEpoch); + liveCandle = refreshedWindow.liveCandleOptional(); } } finally { lock.unlock(); } } else { - log.warn("[CHART-SERVICE] Lock timeout for {}. Proceeding with potential concurrent backfill.", lockKey); - finalizedCandles = backfillAndLoad(symbol, interval, candles, endExclusive); + log.warn( + "[CHART-SERVICE] Lock timeout for {}. " + + "Proceeding with potential concurrent backfill.", + lockKey); + finalizedCandles = backfillAndLoad( + symbol, interval, candles, endExclusive); + OhlcHotWindow refreshedWindow = loadRedisHotWindow(symbol, interval, endEpoch); + liveCandle = refreshedWindow.liveCandleOptional(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -113,7 +141,8 @@ public List show(Long symbolId, OhlcInterval interval, int c // Live Merge: Add the currently updating candle if it's a 'live' request if (to == null) { - return mergeRealTimeCandleIntoSnapshot(finalizedCandles, symbol, base1mBucket, interval); + return mergeRealTimeCandleIntoSnapshot( + finalizedCandles, symbol, endExclusive, interval, liveCandle); } return finalizedCandles; @@ -128,19 +157,31 @@ public void warmUp(Symbol symbol, OhlcInterval interval) { LocalDateTime endExclusive = interval.resolveBucketStart(nowBucket); log.info("[CHART-SERVICE] Warming up cache for {} {}", symbol.getSymbol(), interval); - backfillAndLoad(symbol, interval, ChartCacheConstants.MAX_HOT_WINDOW_SIZE, endExclusive); + backfillAndLoad(symbol, interval, OhlcWindowPolicy.MAX_SIZE, endExclusive); + loadRedisHotWindow( + symbol, + interval, + endExclusive.toEpochSecond(ZoneOffset.UTC) + ); } /** * Loads candles from DB and hydrates the Redis Global Window for future requests. */ - private List backfillAndLoad(Symbol symbol, OhlcInterval interval, int count, LocalDateTime endExclusive) { + private List backfillAndLoad( + Symbol symbol, + OhlcInterval interval, + int count, + LocalDateTime endExclusive + ) { // Backfill a larger portion (MAX_HOT_WINDOW_SIZE) to prevent frequent misses - List hotWindowData = loadFromDb(symbol.getId(), interval, ChartCacheConstants.MAX_HOT_WINDOW_SIZE, endExclusive); + List hotWindowData = loadFromDb( + symbol.getId(), interval, OhlcWindowPolicy.MAX_SIZE, endExclusive); if (!hotWindowData.isEmpty()) { ohlcWindowRepository.saveAll(symbol.getSymbol(), interval.name(), hotWindowData); - ohlcWindowRepository.trim(symbol.getSymbol(), interval.name(), ChartCacheConstants.MAX_HOT_WINDOW_SIZE); + ohlcWindowRepository.trim( + symbol.getSymbol(), interval.name(), OhlcWindowPolicy.MAX_SIZE); } // Return only the requested amount @@ -148,13 +189,21 @@ private List backfillAndLoad(Symbol symbol, OhlcInterval int return hotWindowData.subList(Math.max(0, size - count), size); } - private List loadFromDb(Long symbolId, OhlcInterval interval, int candles, LocalDateTime endExclusive) { + private List loadFromDb( + Long symbolId, + OhlcInterval interval, + int candles, + LocalDateTime endExclusive + ) { LocalDateTime startInclusive = endExclusive.minus(interval.duration().multipliedBy(candles)); return switch (interval) { - case M1 -> toSnapshots(ohlc1mService.findCandlesInBucketRange(symbolId, startInclusive, endExclusive)); - case M5 -> toSnapshots(ohlc5mService.findCandlesInBucketRange(symbolId, startInclusive, endExclusive)); - case M30 -> toSnapshots(ohlc30mService.findCandlesInBucketRange(symbolId, startInclusive, endExclusive)); + case M1 -> toSnapshots(ohlc1mService.findCandlesInBucketRange( + symbolId, startInclusive, endExclusive)); + case M5 -> toSnapshots(ohlc5mService.findCandlesInBucketRange( + symbolId, startInclusive, endExclusive)); + case M30 -> toSnapshots(ohlc30mService.findCandlesInBucketRange( + symbolId, startInclusive, endExclusive)); }; } @@ -166,15 +215,16 @@ private List toSnapshots(List candle private List mergeRealTimeCandleIntoSnapshot( List snapshots, Symbol symbol, - LocalDateTime baseBucket, OhlcInterval interval) { + LocalDateTime baseBucket, OhlcInterval interval, + Optional cachedLiveCandle) { - if (liveKlineRepository.isEmpty()) { - return snapshots; + Optional liveKlineOpt = cachedLiveCandle; + if (liveKlineOpt.isEmpty() && hotWindowStore.get(symbol.getSymbol(), interval.name()).isEmpty() + && liveKlineRepository.isPresent()) { + liveKlineOpt = liveKlineRepository.get().findBySymbolAndInterval( + symbol.getSymbol(), interval.name()); } - Optional liveKlineOpt = liveKlineRepository.get().findBySymbolAndInterval( - symbol.getSymbol(), interval.name()); - if (liveKlineOpt.isEmpty()) { return snapshots; } @@ -212,5 +262,35 @@ private List mergeRealTimeCandleIntoSnapshot( return result; } + + private OhlcHotWindow loadRedisHotWindow(Symbol symbol, OhlcInterval interval, long endEpoch) { + long expectedVersion = hotWindowStore.eventVersion(symbol.getSymbol(), interval.name()); + List finalized = ohlcWindowRepository.findRange( + symbol.getSymbol(), + interval.name(), + endEpoch, + OhlcWindowPolicy.MAX_SIZE + ); + Optional live = liveKlineRepository.flatMap(repository -> + repository.findBySymbolAndInterval(symbol.getSymbol(), interval.name())); + hotWindowStore.replaceIfVersion( + symbol.getSymbol(), + interval.name(), + finalized, + live, + Instant.now(clock), + expectedVersion + ); + return hotWindowStore.get(symbol.getSymbol(), interval.name()) + .orElseGet(() -> new OhlcHotWindow(finalized, live.orElse(null), Instant.now(clock), 0)); + } + + private boolean isFresh(OhlcHotWindow window) { + if (Instant.EPOCH.equals(window.synchronizedAt())) { + return false; + } + long ageMillis = Duration.between(window.synchronizedAt(), clock.instant()).toMillis(); + return ageMillis >= 0 && ageMillis <= ChartCacheConstants.HOT_WINDOW_STALE_AFTER_MILLIS; + } } diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcHotWindowRefreshService.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcHotWindowRefreshService.java new file mode 100644 index 00000000..7525b897 --- /dev/null +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcHotWindowRefreshService.java @@ -0,0 +1,107 @@ +package com.coinflow.chart.service.sync; + +import com.coinflow.chart.cache.hot.OhlcHotWindowStore; +import com.coinflow.domain.ohlc.constant.OhlcInterval; +import com.coinflow.domain.ohlc.constant.OhlcWindowPolicy; +import com.coinflow.domain.ohlc.repository.LiveKlineRepository; +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.domain.symbol.domain.Symbol; +import com.coinflow.domain.symbol.service.SymbolService; +import com.coinflow.event.kline.KlineEvent; +import com.coinflow.util.TimeBucket; +import java.time.Clock; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Optional; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +@Slf4j +@Service +@RequiredArgsConstructor +@ConditionalOnProperty( + name = "coinflow.chart.hot-window.enabled", + havingValue = "true", + matchIfMissing = true +) +public class OhlcHotWindowRefreshService { + + private final OhlcWindowRepository ohlcWindowRepository; + private final LiveKlineRepository liveKlineRepository; + private final OhlcHotWindowStore hotWindowStore; + private final SymbolService symbolService; + private final Clock clock; + + private volatile List symbols = List.of(); + + @EventListener(ApplicationReadyEvent.class) + public void loadSymbols() { + refreshSymbols(); + } + + @Scheduled( + fixedDelayString = "${coinflow.chart.hot-window.symbol-refresh-interval-ms:60000}", + initialDelayString = "${coinflow.chart.hot-window.symbol-refresh-interval-ms:60000}" + ) + public void refreshSymbols() { + try { + symbols = List.copyOf(symbolService.findAll()); + } catch (Exception e) { + log.error("[HOT-WINDOW] Failed to refresh symbols; keeping the previous symbol list", e); + } + } + + @Scheduled( + fixedDelayString = "${coinflow.chart.hot-window.refresh-interval-ms:1000}", + initialDelayString = "${coinflow.chart.hot-window.initial-delay-ms:1000}" + ) + public void refreshAll() { + for (Symbol symbol : symbols) { + for (OhlcInterval interval : OhlcInterval.values()) { + refresh(symbol, interval); + } + } + } + + public void refresh(Symbol symbol, OhlcInterval interval) { + try { + long expectedVersion = hotWindowStore.eventVersion( + symbol.getSymbol(), interval.name()); + LocalDateTime currentBucket = TimeBucket.to1m(clock.instant()); + long endExclusive = interval.resolveBucketStart(currentBucket) + .toEpochSecond(ZoneOffset.UTC); + List finalizedCandles = ohlcWindowRepository.findRange( + symbol.getSymbol(), + interval.name(), + endExclusive, + OhlcWindowPolicy.MAX_SIZE + ); + Optional liveCandle = liveKlineRepository.findBySymbolAndInterval( + symbol.getSymbol(), interval.name()); + + boolean replaced = hotWindowStore.replaceIfVersion( + symbol.getSymbol(), + interval.name(), + finalizedCandles, + liveCandle, + Instant.now(clock), + expectedVersion + ); + if (!replaced) { + log.trace("[HOT-WINDOW] Skipped stale poll result for {} {}", + symbol.getSymbol(), interval); + } + } catch (Exception e) { + log.error("[HOT-WINDOW] Refresh failed for {} {}; keeping the previous snapshot", + symbol.getSymbol(), interval, e); + } + } +} diff --git a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcWindowSyncService.java b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcWindowSyncService.java index f49747ba..d6f6b1da 100644 --- a/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcWindowSyncService.java +++ b/backend/coinflow-api-app/src/main/java/com/coinflow/chart/service/sync/OhlcWindowSyncService.java @@ -1,11 +1,10 @@ package com.coinflow.chart.service.sync; -import com.coinflow.chart.constant.ChartCacheConstants; -import com.coinflow.chart.repository.RedisOhlcWindowRepository; -import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.chart.cache.hot.OhlcHotWindowStore; import com.coinflow.event.kline.KlineEvent; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import java.nio.charset.StandardCharsets; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.connection.Message; @@ -14,39 +13,26 @@ import org.springframework.lang.Nullable; import org.springframework.stereotype.Service; -import java.time.LocalDateTime; -import java.time.ZoneOffset; - /** - * Service that listens for kline events and updates the Redis-based sliding window. - * Ensures the chart cache remains consistent with the latest finalized candles. + * Applies Pub/Sub events to the local hot window between periodic Redis reconciliations. */ @Slf4j @Service @RequiredArgsConstructor public class OhlcWindowSyncService implements MessageListener { - private final RedisOhlcWindowRepository ohlcWindowRepository; + private final OhlcHotWindowStore hotWindowStore; private final ObjectMapper objectMapper; @Override public void onMessage(@NonNull Message message, @Nullable byte[] pattern) { try { - String json = new String(message.getBody()); + String json = new String(message.getBody(), StandardCharsets.UTF_8); KlineEvent event = objectMapper.readValue(json, KlineEvent.class); - // Our cache only stores finalized (closed) candles for pure data integrity. - // Late ticks are also marked as closed: true, ensuring they update the ZSET. - if (event.closed()) { - OhlcCandleSnapshot snapshot = toSnapshot(event); - ohlcWindowRepository.save(event.symbol(), event.interval(), snapshot); - - // Maintain the sliding window size using shared configuration - ohlcWindowRepository.trim(event.symbol(), event.interval(), ChartCacheConstants.MAX_HOT_WINDOW_SIZE); - - log.debug("[CHART-SYNC] Synchronized closed candle for {} {}: {}", - event.symbol(), event.interval(), snapshot.bucketTime()); - } + hotWindowStore.applyEvent(event); + log.trace("[CHART-SYNC] Applied event to local window for {} {}: {}", + event.symbol(), event.interval(), event.startTime()); } catch (JsonProcessingException e) { log.error("Failed to deserialize KlineEvent for chart synchronization", e); @@ -55,16 +41,4 @@ public void onMessage(@NonNull Message message, @Nullable byte[] pattern) { } } - private OhlcCandleSnapshot toSnapshot(KlineEvent event) { - LocalDateTime bucketTime = LocalDateTime.ofEpochSecond(event.startTime(), 0, ZoneOffset.UTC); - return new OhlcCandleSnapshot( - bucketTime, - event.startTime(), - event.open(), - event.high(), - event.low(), - event.close(), - event.volume() - ); - } } diff --git a/backend/coinflow-api-app/src/main/resources/application-api.yml b/backend/coinflow-api-app/src/main/resources/application-api.yml index ddb0af43..a62a384d 100644 --- a/backend/coinflow-api-app/src/main/resources/application-api.yml +++ b/backend/coinflow-api-app/src/main/resources/application-api.yml @@ -32,6 +32,14 @@ logging: server: port: 8080 +coinflow: + chart: + hot-window: + enabled: true + refresh-interval-ms: 1000 + initial-delay-ms: 1000 + symbol-refresh-interval-ms: 60000 + management: endpoints: web: diff --git a/backend/coinflow-api-app/src/test/java/com/coinflow/chart/cache/hot/OhlcHotWindowStoreTest.java b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/cache/hot/OhlcHotWindowStoreTest.java new file mode 100644 index 00000000..fba5ba52 --- /dev/null +++ b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/cache/hot/OhlcHotWindowStoreTest.java @@ -0,0 +1,108 @@ +package com.coinflow.chart.cache.hot; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.event.kline.KlineEvent; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.math.BigDecimal; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Optional; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class OhlcHotWindowStoreTest { + + private OhlcHotWindowStore store; + + @BeforeEach + void setUp() { + store = new OhlcHotWindowStore(new SimpleMeterRegistry()); + } + + @Test + void replacesTheWholeWindowAndReturnsTheRequestedTail() { + KlineEvent live = event(180, "103", false); + store.replace( + "btcusdt", + "M1", + List.of(snapshot(60, "101"), snapshot(120, "102")), + Optional.of(live), + Instant.parse("2026-08-14T00:03:00Z") + ); + + OhlcHotWindow window = store.get("BTCUSDT", "M1").orElseThrow(); + + assertEquals(List.of(snapshot(120, "102")), window.findFinalizedRange(180, 1)); + assertEquals(live, window.liveCandle()); + } + + @Test + void doesNotOverwriteAnEventThatArrivedDuringPolling() { + store.replace( + "btcusdt", "M1", List.of(snapshot(60, "101")), Optional.empty(), Instant.now()); + long versionBeforePoll = store.eventVersion("btcusdt", "M1"); + + KlineEvent newerLive = event(120, "105", false); + store.applyEvent(newerLive); + + boolean replaced = store.replaceIfVersion( + "btcusdt", + "M1", + List.of(snapshot(60, "101")), + Optional.of(event(120, "102", false)), + Instant.now(), + versionBeforePoll + ); + + assertFalse(replaced); + assertEquals(newerLive, store.get("btcusdt", "M1").orElseThrow().liveCandle()); + } + + @Test + void closedEventMovesTheLiveCandleIntoTheFinalizedWindow() { + store.applyEvent(event(120, "102", false)); + store.applyEvent(event(120, "104", true)); + + OhlcHotWindow window = store.get("btcusdt", "M1").orElseThrow(); + + assertTrue(window.liveCandleOptional().isEmpty()); + assertEquals("104", window.finalizedCandles().get(0).closePrice().toPlainString()); + } + + private OhlcCandleSnapshot snapshot(long epochSeconds, String close) { + LocalDateTime bucket = LocalDateTime.ofEpochSecond(epochSeconds, 0, ZoneOffset.UTC); + BigDecimal closePrice = new BigDecimal(close); + return new OhlcCandleSnapshot( + bucket, + epochSeconds, + closePrice, + closePrice, + closePrice, + closePrice, + BigDecimal.ONE + ); + } + + private KlineEvent event(long startTime, String close, boolean closed) { + BigDecimal closePrice = new BigDecimal(close); + return KlineEvent.builder() + .symbol("btcusdt") + .interval("M1") + .startTime(startTime) + .closeTime(startTime + 59) + .open(closePrice) + .high(closePrice) + .low(closePrice) + .close(closePrice) + .volume(BigDecimal.ONE) + .trades(1) + .closed(closed) + .build(); + } +} diff --git a/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/OhlcChartServiceTest.java b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/OhlcChartServiceTest.java new file mode 100644 index 00000000..cf779bbc --- /dev/null +++ b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/OhlcChartServiceTest.java @@ -0,0 +1,136 @@ +package com.coinflow.chart.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.coinflow.chart.cache.hot.OhlcHotWindow; +import com.coinflow.chart.cache.hot.OhlcHotWindowStore; +import com.coinflow.domain.ohlc.cache.OhlcChartStore; +import com.coinflow.domain.ohlc.constant.OhlcInterval; +import com.coinflow.domain.ohlc.repository.LiveKlineRepository; +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; +import com.coinflow.domain.ohlc.service.Ohlc1mService; +import com.coinflow.domain.ohlc.service.Ohlc30mService; +import com.coinflow.domain.ohlc.service.Ohlc5mService; +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.domain.symbol.domain.Symbol; +import com.coinflow.domain.symbol.service.SymbolService; +import com.coinflow.event.kline.KlineEvent; +import java.math.BigDecimal; +import java.time.Clock; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Optional; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class OhlcChartServiceTest { + + @Mock + private OhlcChartStore chartStore; + + @Mock + private OhlcWindowRepository ohlcWindowRepository; + + @Mock + private OhlcHotWindowStore hotWindowStore; + + @Mock + private Ohlc1mService ohlc1mService; + + @Mock + private Ohlc5mService ohlc5mService; + + @Mock + private Ohlc30mService ohlc30mService; + + @Mock + private LiveKlineRepository liveKlineRepository; + + @Mock + private SymbolService symbolService; + + private Clock clock; + private OhlcChartService service; + private Symbol symbol; + + @BeforeEach + void setUp() { + clock = Clock.fixed(Instant.parse("2026-08-14T12:03:30Z"), ZoneOffset.UTC); + service = new OhlcChartService( + chartStore, + ohlcWindowRepository, + hotWindowStore, + clock, + ohlc1mService, + ohlc5mService, + ohlc30mService, + Optional.of(liveKlineRepository), + symbolService + ); + symbol = Symbol.builder().id(1L).symbol("btcusdt").build(); + } + + @Test + void freshHotWindowServesFinalizedAndLiveCandlesWithoutRedis() { + long first = Instant.parse("2026-08-14T12:01:00Z").getEpochSecond(); + long second = Instant.parse("2026-08-14T12:02:00Z").getEpochSecond(); + long liveStart = Instant.parse("2026-08-14T12:03:00Z").getEpochSecond(); + KlineEvent live = event(liveStart, "103"); + OhlcHotWindow window = new OhlcHotWindow( + List.of(snapshot(first, "101"), snapshot(second, "102")), + live, + clock.instant(), + 0 + ); + + when(symbolService.findSymbol(1L)).thenReturn(symbol); + when(hotWindowStore.get("btcusdt", "M1")).thenReturn(Optional.of(window)); + + List result = service.show(1L, OhlcInterval.M1, 2, null); + + assertEquals(3, result.size()); + assertEquals(liveStart, result.get(2).epochSeconds()); + assertEquals("103", result.get(2).closePrice().toPlainString()); + verify(ohlcWindowRepository, never()).findRange("btcusdt", "M1", second + 60, 1000); + verify(liveKlineRepository, never()).findBySymbolAndInterval("btcusdt", "M1"); + } + + private OhlcCandleSnapshot snapshot(long epochSeconds, String close) { + BigDecimal value = new BigDecimal(close); + return new OhlcCandleSnapshot( + LocalDateTime.ofEpochSecond(epochSeconds, 0, ZoneOffset.UTC), + epochSeconds, + value, + value, + value, + value, + BigDecimal.ONE + ); + } + + private KlineEvent event(long startTime, String close) { + BigDecimal value = new BigDecimal(close); + return KlineEvent.builder() + .symbol("btcusdt") + .interval("M1") + .startTime(startTime) + .closeTime(startTime + 59) + .open(value) + .high(value) + .low(value) + .close(value) + .volume(BigDecimal.ONE) + .trades(1) + .closed(false) + .build(); + } +} diff --git a/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/sync/OhlcHotWindowRefreshServiceTest.java b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/sync/OhlcHotWindowRefreshServiceTest.java new file mode 100644 index 00000000..3291f318 --- /dev/null +++ b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/sync/OhlcHotWindowRefreshServiceTest.java @@ -0,0 +1,86 @@ +package com.coinflow.chart.service.sync; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.coinflow.chart.cache.hot.OhlcHotWindowStore; +import com.coinflow.domain.ohlc.constant.OhlcInterval; +import com.coinflow.domain.ohlc.repository.LiveKlineRepository; +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.coinflow.domain.symbol.domain.Symbol; +import com.coinflow.domain.symbol.service.SymbolService; +import java.math.BigDecimal; +import java.time.Clock; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Optional; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class OhlcHotWindowRefreshServiceTest { + + @Mock + private OhlcWindowRepository ohlcWindowRepository; + + @Mock + private LiveKlineRepository liveKlineRepository; + + @Mock + private OhlcHotWindowStore hotWindowStore; + + @Mock + private SymbolService symbolService; + + @Test + void refreshReadsTheEntireRedisWindowAndAtomicallyReplacesCaffeine() { + Clock clock = Clock.fixed(Instant.parse("2026-08-14T12:03:30Z"), ZoneOffset.UTC); + OhlcHotWindowRefreshService service = new OhlcHotWindowRefreshService( + ohlcWindowRepository, + liveKlineRepository, + hotWindowStore, + symbolService, + clock + ); + Symbol symbol = Symbol.builder().id(1L).symbol("btcusdt").build(); + long endExclusive = Instant.parse("2026-08-14T12:03:00Z").getEpochSecond(); + List finalized = List.of(snapshot(endExclusive - 60)); + + when(hotWindowStore.eventVersion("btcusdt", "M1")).thenReturn(7L); + when(ohlcWindowRepository.findRange("btcusdt", "M1", endExclusive, 1000)) + .thenReturn(finalized); + when(liveKlineRepository.findBySymbolAndInterval("btcusdt", "M1")) + .thenReturn(Optional.empty()); + + service.refresh(symbol, OhlcInterval.M1); + + verify(hotWindowStore).replaceIfVersion( + eq("btcusdt"), + eq("M1"), + eq(finalized), + eq(Optional.empty()), + any(Instant.class), + eq(7L) + ); + } + + private OhlcCandleSnapshot snapshot(long epochSeconds) { + LocalDateTime bucket = LocalDateTime.ofEpochSecond(epochSeconds, 0, ZoneOffset.UTC); + return new OhlcCandleSnapshot( + bucket, + epochSeconds, + BigDecimal.ONE, + BigDecimal.ONE, + BigDecimal.ONE, + BigDecimal.ONE, + BigDecimal.ONE + ); + } +} diff --git a/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/sync/OhlcWindowSyncServiceTest.java b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/sync/OhlcWindowSyncServiceTest.java new file mode 100644 index 00000000..e80d54ea --- /dev/null +++ b/backend/coinflow-api-app/src/test/java/com/coinflow/chart/service/sync/OhlcWindowSyncServiceTest.java @@ -0,0 +1,48 @@ +package com.coinflow.chart.service.sync; + +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.coinflow.chart.cache.hot.OhlcHotWindowStore; +import com.coinflow.event.kline.KlineEvent; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.math.BigDecimal; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.data.redis.connection.Message; + +@ExtendWith(MockitoExtension.class) +class OhlcWindowSyncServiceTest { + + @Mock + private OhlcHotWindowStore hotWindowStore; + + @Mock + private Message message; + + @Test + void pubSubEventUpdatesOnlyTheLocalHotWindow() throws Exception { + ObjectMapper objectMapper = new ObjectMapper(); + OhlcWindowSyncService service = new OhlcWindowSyncService(hotWindowStore, objectMapper); + KlineEvent event = KlineEvent.builder() + .symbol("btcusdt") + .interval("M1") + .startTime(60) + .closeTime(119) + .open(BigDecimal.ONE) + .high(BigDecimal.TEN) + .low(BigDecimal.ONE) + .close(BigDecimal.TEN) + .volume(BigDecimal.ONE) + .trades(1) + .closed(false) + .build(); + when(message.getBody()).thenReturn(objectMapper.writeValueAsBytes(event)); + + service.onMessage(message, null); + + verify(hotWindowStore).applyEvent(event); + } +} diff --git a/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/infrastructure/persistence/DbPersistService.java b/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/infrastructure/persistence/DbPersistService.java index e1ad712c..bd231e55 100644 --- a/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/infrastructure/persistence/DbPersistService.java +++ b/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/infrastructure/persistence/DbPersistService.java @@ -85,9 +85,9 @@ private void saveByInterval(Symbol symbol, LocalDateTime bucketTime, ClosedKline */ @Recover public CompletableFuture recover(Exception e, String symbolCode, ClosedKlineSnapshot closedSnapshot) { - log.error("[DB-ASYNC-FATAL] All retry attempts failed for {} {} candle. Batch will reconcile later. error={}", + log.error("[DB-ASYNC-FATAL] All retry attempts failed for {} {} candle. Message will remain pending. error={}", symbolCode, closedSnapshot.interval(), e.getMessage()); - return CompletableFuture.completedFuture(null); + return CompletableFuture.failedFuture(e); } } diff --git a/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/TickProcessService.java b/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/TickProcessService.java index 6da41ac7..b0a2d8fd 100644 --- a/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/TickProcessService.java +++ b/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/TickProcessService.java @@ -7,12 +7,17 @@ import com.coinflow.domain.aggregation.domain.vo.ClosedKlineSnapshot; import com.coinflow.domain.aggregation.domain.vo.KlineSnapshot; import com.coinflow.domain.aggregation.service.KlineAggregatorService; +import com.coinflow.domain.ohlc.constant.OhlcWindowPolicy; import com.coinflow.domain.ohlc.repository.LiveKlineRepository; +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; import com.coinflow.event.kline.KlineEvent; import com.coinflow.monitoring.MetricRecorder; import com.fasterxml.jackson.databind.ObjectMapper; import java.math.BigDecimal; import java.time.Duration; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -36,6 +41,7 @@ public class TickProcessService { private final KlineAggregatorService klineAggregatorService; private final LiveKlineRepository liveKlineRepository; + private final OhlcWindowRepository ohlcWindowRepository; private final KlineBroadcaster klineBroadcaster; private final TickerBroadcaster tickerBroadcaster; private final DbPersistService dbPersistService; @@ -87,6 +93,7 @@ public void process(String symbol, BigDecimal price, BigDecimal quantity, long e completeAndAcknowledge(dbFutures, streamKey, group, recordId, symbol, startNanos); } catch (Exception e) { + processedIdCache.invalidate(recordId.getValue()); log.error("[Consumer] Critical failure processing tick - symbol={}", symbol, e); recordFailure(symbol); throw e; @@ -130,8 +137,40 @@ private void processCandidate(String symbol, ClosedKlineSnapshot snapshot) { private void processFinalizedCandidate(String symbol, ClosedKlineSnapshot snapshot, List> futures) { - processCandidate(symbol, snapshot); - futures.add(dbPersistService.persistClosedCandleAsync(symbol, snapshot)); + KlineEvent event = toEvent(symbol, snapshot.interval(), snapshot.snapshot()); + String json; + try { + json = objectMapper.writeValueAsString(event); + } catch (Exception e) { + throw new IllegalStateException("Failed to serialize finalized kline event", e); + } + + CompletableFuture finalizedFuture = dbPersistService + .persistClosedCandleAsync(symbol, snapshot) + .thenRun(() -> { + OhlcCandleSnapshot candle = toOhlcSnapshot(snapshot.snapshot()); + ohlcWindowRepository.save(symbol, snapshot.interval(), candle); + ohlcWindowRepository.trim( + symbol, snapshot.interval(), OhlcWindowPolicy.MAX_SIZE); + liveKlineRepository.deleteIfStartTimeMatches( + symbol, snapshot.interval(), snapshot.snapshot().startTime()); + klineBroadcaster.broadcast(event, json); + }); + futures.add(finalizedFuture); + } + + private OhlcCandleSnapshot toOhlcSnapshot(KlineSnapshot snapshot) { + LocalDateTime bucketTime = LocalDateTime.ofEpochSecond( + snapshot.startTime(), 0, ZoneOffset.UTC); + return new OhlcCandleSnapshot( + bucketTime, + snapshot.startTime(), + snapshot.open(), + snapshot.high(), + snapshot.low(), + snapshot.close(), + snapshot.volume() + ); } private KlineEvent toEvent(String symbol, String interval, KlineSnapshot snapshot) { @@ -159,6 +198,7 @@ private void completeAndAcknowledge(List> futures, CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .thenRun(() -> finalizeProcess(streamKey, group, recordId, symbol, startNanos)) .exceptionally(ex -> { + processedIdCache.invalidate(recordId.getValue()); log.error("Async pipeline failed for {}. Message will stick in PENDING.", symbol, ex); recordFailure(symbol); return (Void) null; diff --git a/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/TickProcessServiceTest.java b/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/TickProcessServiceTest.java index 688193ed..c788137b 100644 --- a/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/TickProcessServiceTest.java +++ b/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/TickProcessServiceTest.java @@ -5,6 +5,7 @@ import com.coinflow.domain.aggregation.domain.vo.AggregationResult; import com.coinflow.domain.aggregation.service.KlineAggregatorService; import com.coinflow.domain.ohlc.repository.LiveKlineRepository; +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; import com.coinflow.aggregation.infrastructure.persistence.DbPersistService; import com.coinflow.event.kline.KlineEvent; import com.coinflow.monitoring.MetricRecorder; @@ -42,6 +43,8 @@ class TickProcessServiceTest { @Mock private LiveKlineRepository liveKlineRepository; @Mock + private OhlcWindowRepository ohlcWindowRepository; + @Mock private KlineBroadcaster klineBroadcaster; @Mock private TickerBroadcaster tickerBroadcaster; @@ -103,7 +106,11 @@ void processLateTickTest() { // 1. Ticker 최신성 기반 전파 확인 () -> verify(tickerBroadcaster, times(1)).broadcast(anyString()), // 2. 캐시 저장 및 브로드캐스트 전파 확인 - () -> verify(liveKlineRepository, times(1)).save(any(KlineEvent.class), anyString()), + () -> verify(liveKlineRepository, never()).save(any(KlineEvent.class), anyString()), + () -> verify(ohlcWindowRepository, times(1)).save(eq(symbol), eq("M1"), any()), + () -> verify(ohlcWindowRepository, times(1)).trim(eq(symbol), eq("M1"), eq(1000)), + () -> verify(liveKlineRepository, times(1)) + .deleteIfStartTimeMatches(symbol, "M1", lateSnapshot.startTime()), () -> verify(klineBroadcaster, times(1)).broadcast(any(KlineEvent.class), anyString()), // 3. 메인 스레드 점유 시간(나노초) 기록 확인 () -> verify(metricRecorder, atLeastOnce()).recordTimeNanos(eq(TICK_MAIN_THREAD_LATENCY), @@ -114,4 +121,31 @@ void processLateTickTest() { () -> verify(batchAckWorker, timeout(1000)).addAck(recordId) ); } + + @Test + @DisplayName("Closed candle updates Redis and Pub/Sub only after DB persistence succeeds") + void finalizedCandleWaitsForDatabaseBeforePublishing() { + KlineSnapshot finalizedSnapshot = new KlineSnapshot( + 120L, 179L, price, price, price, price, quantity, 1, true); + ClosedKlineSnapshot closed = new ClosedKlineSnapshot("M1", finalizedSnapshot); + AggregationResult result = new AggregationResult(List.of(closed), List.of(), List.of()); + CompletableFuture dbFuture = new CompletableFuture<>(); + + when(klineAggregatorService.processTickAndGetResult( + eq(symbol), eq(price), eq(quantity), eq(eventTime))).thenReturn(result); + when(dbPersistService.persistClosedCandleAsync(symbol, closed)).thenReturn(dbFuture); + + tickProcessService.process( + symbol, price, quantity, eventTime, "mystream", "mygroup", RecordId.of("124-0")); + + verify(ohlcWindowRepository, never()).save(anyString(), anyString(), any()); + verify(klineBroadcaster, never()).broadcast(any(), anyString()); + verify(batchAckWorker, never()).addAck(any()); + + dbFuture.complete(null); + + verify(ohlcWindowRepository).save(eq(symbol), eq("M1"), any()); + verify(klineBroadcaster).broadcast(any(KlineEvent.class), anyString()); + verify(batchAckWorker, timeout(1000)).addAck(RecordId.of("124-0")); + } } diff --git a/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/constant/OhlcWindowPolicy.java b/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/constant/OhlcWindowPolicy.java new file mode 100644 index 00000000..ebac02fe --- /dev/null +++ b/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/constant/OhlcWindowPolicy.java @@ -0,0 +1,10 @@ +package com.coinflow.domain.ohlc.constant; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class OhlcWindowPolicy { + + public static final int MAX_SIZE = 1000; +} diff --git a/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/LiveKlineRepository.java b/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/LiveKlineRepository.java index dacea76d..cda421f3 100644 --- a/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/LiveKlineRepository.java +++ b/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/LiveKlineRepository.java @@ -24,4 +24,9 @@ public interface LiveKlineRepository { */ void delete(String symbol, String interval); + /** + * Deletes the live value only when it still represents the supplied bucket. + */ + void deleteIfStartTimeMatches(String symbol, String interval, long startTime); + } diff --git a/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/OhlcWindowRepository.java b/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/OhlcWindowRepository.java new file mode 100644 index 00000000..45977a9f --- /dev/null +++ b/backend/coinflow-core/src/main/java/com/coinflow/domain/ohlc/repository/OhlcWindowRepository.java @@ -0,0 +1,18 @@ +package com.coinflow.domain.ohlc.repository; + +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import java.util.List; + +/** + * Shared finalized-candle window backed by Redis. + */ +public interface OhlcWindowRepository { + + void save(String symbol, String interval, OhlcCandleSnapshot snapshot); + + void saveAll(String symbol, String interval, List snapshots); + + List findRange(String symbol, String interval, long to, int limit); + + void trim(String symbol, String interval, int limit); +} diff --git a/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/LiveKlineRepositoryImpl.java b/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/LiveKlineRepositoryImpl.java index 448cb57f..f2d54a7a 100644 --- a/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/LiveKlineRepositoryImpl.java +++ b/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/LiveKlineRepositoryImpl.java @@ -8,6 +8,7 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Repository; @Slf4j @@ -16,6 +17,15 @@ public class LiveKlineRepositoryImpl implements LiveKlineRepository { public static final String KEY_PREFIX = "kline:live:"; + private static final DefaultRedisScript DELETE_IF_BUCKET_MATCHES_SCRIPT = + new DefaultRedisScript<>( + "local value = redis.call('GET', KEYS[1]); " + + "if not value then return 0 end; " + + "local event = cjson.decode(value); " + + "if tostring(event.startTime) == ARGV[1] then " + + "return redis.call('DEL', KEYS[1]); end; return 0;", + Long.class + ); private final StringRedisTemplate redisTemplate; private final ObjectMapper objectMapper; @@ -51,6 +61,16 @@ public void delete(String symbol, String interval) { redisTemplate.delete(key); } + @Override + public void deleteIfStartTimeMatches(String symbol, String interval, long startTime) { + String key = buildKey(symbol, interval); + redisTemplate.execute( + DELETE_IF_BUCKET_MATCHES_SCRIPT, + java.util.List.of(key), + Long.toString(startTime) + ); + } + private String buildKey(String symbol, String interval) { return KEY_PREFIX + symbol.toLowerCase() + ":" + interval; } diff --git a/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/RedisOhlcWindowRepositoryImpl.java b/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/RedisOhlcWindowRepositoryImpl.java new file mode 100644 index 00000000..f7e5ba7b --- /dev/null +++ b/backend/coinflow-infra-redis/src/main/java/com/coinflow/aggregation/repository/RedisOhlcWindowRepositoryImpl.java @@ -0,0 +1,128 @@ +package com.coinflow.aggregation.repository; + +import com.coinflow.domain.ohlc.repository.OhlcWindowRepository; +import com.coinflow.domain.ohlc.snapshot.OhlcCandleSnapshot; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Repository; + +@Slf4j +@Repository +@RequiredArgsConstructor +public class RedisOhlcWindowRepositoryImpl implements OhlcWindowRepository { + + private static final String KEY_PREFIX = "klines:window:"; + private static final DefaultRedisScript UPSERT_SCRIPT = new DefaultRedisScript<>( + "redis.call('ZREMRANGEBYSCORE', KEYS[1], ARGV[1], ARGV[1]); " + + "return redis.call('ZADD', KEYS[1], ARGV[1], ARGV[2]);", + Long.class + ); + private static final DefaultRedisScript REPLACE_RANGE_SCRIPT = new DefaultRedisScript<>( + "redis.call('ZREMRANGEBYSCORE', KEYS[1], ARGV[1], ARGV[2]); " + + "local added = 0; " + + "for i = 3, #ARGV, 2 do " + + "added = added + redis.call('ZADD', KEYS[1], ARGV[i], ARGV[i + 1]); end; " + + "return added;", + Long.class + ); + + private final RedisTemplate redisTemplate; + private final ObjectMapper objectMapper; + + @Override + public void save(String symbol, String interval, OhlcCandleSnapshot snapshot) { + String key = buildKey(symbol, interval); + String score = Long.toString(snapshot.epochSeconds()); + String json = serialize(snapshot); + + redisTemplate.execute(UPSERT_SCRIPT, List.of(key), score, json); + log.trace("[REDIS-WINDOW] Upserted candle for {} {}: {}", + symbol, interval, snapshot.bucketTime()); + } + + @Override + public void saveAll(String symbol, String interval, List snapshots) { + if (snapshots == null || snapshots.isEmpty()) { + return; + } + + String key = buildKey(symbol, interval); + long minScore = snapshots.stream() + .mapToLong(OhlcCandleSnapshot::epochSeconds) + .min() + .orElseThrow(); + long maxScore = snapshots.stream() + .mapToLong(OhlcCandleSnapshot::epochSeconds) + .max() + .orElseThrow(); + + List args = new ArrayList<>(2 + snapshots.size() * 2); + args.add(Long.toString(minScore)); + args.add(Long.toString(maxScore)); + for (OhlcCandleSnapshot snapshot : snapshots) { + args.add(Long.toString(snapshot.epochSeconds())); + args.add(serialize(snapshot)); + } + redisTemplate.execute(REPLACE_RANGE_SCRIPT, List.of(key), args.toArray()); + + log.debug("[REDIS-WINDOW] Batch saved {} candles for {} {}", + snapshots.size(), symbol, interval); + } + + @Override + public List findRange(String symbol, String interval, long to, int limit) { + Set jsonSet = redisTemplate.opsForZSet() + .reverseRangeByScore(buildKey(symbol, interval), -1, (double) to - 1, 0, limit); + + if (jsonSet == null || jsonSet.isEmpty()) { + return Collections.emptyList(); + } + + List results = jsonSet.stream() + .map(this::deserialize) + .filter(Objects::nonNull) + .toList(); + List ascending = new ArrayList<>(results); + Collections.reverse(ascending); + return ascending; + } + + @Override + public void trim(String symbol, String interval, int limit) { + String key = buildKey(symbol, interval); + Long size = redisTemplate.opsForZSet().size(key); + if (size != null && size > limit) { + redisTemplate.opsForZSet().removeRange(key, 0, size - limit - 1); + } + } + + private String buildKey(String symbol, String interval) { + return KEY_PREFIX + symbol.toLowerCase() + ":" + interval; + } + + private String serialize(OhlcCandleSnapshot snapshot) { + try { + return objectMapper.writeValueAsString(snapshot); + } catch (JsonProcessingException e) { + throw new IllegalStateException("Failed to serialize OHLC window snapshot", e); + } + } + + private OhlcCandleSnapshot deserialize(String json) { + try { + return objectMapper.readValue(json, OhlcCandleSnapshot.class); + } catch (JsonProcessingException e) { + log.error("Failed to deserialize OHLC window snapshot: {}", json, e); + return null; + } + } +} diff --git a/frontend/src/components/Chart/TradingChart.tsx b/frontend/src/components/Chart/TradingChart.tsx index 0b607386..fad1f6b7 100644 --- a/frontend/src/components/Chart/TradingChart.tsx +++ b/frontend/src/components/Chart/TradingChart.tsx @@ -134,21 +134,31 @@ export const TradingChart = () => { const candleTime = msg.startTime as number; const isHistorical = candleTime < lastCandleTimeRef.current; + const liveCandle: ChartCandle = { + time: candleTime as Time, + open: msg.open, + high: msg.high, + low: msg.low, + close: msg.close, + }; + const liveVolume: VolumeBar = { + time: candleTime as Time, + value: msg.volume, + color: msg.close >= msg.open + ? CHART_COLORS.UP_TRANSPARENT + : CHART_COLORS.DOWN_TRANSPARENT, + }; + + // Keep WebSocket data in the merge source so a slower REST response + // cannot overwrite a newer value for the same candle timestamp. + rawDataRef.current = { + candles: uniqueSortData([...rawDataRef.current.candles, liveCandle]), + volumes: uniqueSortData([...rawDataRef.current.volumes, liveVolume]), + }; try { - mainSeriesRef.current.update({ - time: candleTime as Time, - open: msg.open, - high: msg.high, - low: msg.low, - close: msg.close, - }, isHistorical); - - volumeSeriesRef.current.update({ - time: candleTime as Time, - value: msg.volume, - color: msg.close >= msg.open ? CHART_COLORS.UP_TRANSPARENT : CHART_COLORS.DOWN_TRANSPARENT, - }, isHistorical); + mainSeriesRef.current.update(liveCandle, isHistorical); + volumeSeriesRef.current.update(liveVolume, isHistorical); if (!isHistorical) { lastCandleTimeRef.current = Math.max(lastCandleTimeRef.current, candleTime);