Store latest prices into redis cache #2321
Conversation
📝 WalkthroughWalkthroughThe changes in this pull request introduce several new functionalities and improvements across multiple files. A new function Changes
Possibly related PRs
Suggested reviewers
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Outside diff range and nitpick comments (6)
node/pkg/common/types/types.go (3)
86-86: Consider adding context timeout.The method accepts a context but doesn't enforce any timeout. For cache operations, it's important to have timeouts to prevent hanging operations.
-func (m *LatestFeedDataMap) GetLatestFeedDataFromCache(ctx context.Context, feedIds []int32) ([]*FeedData, error) { +func (m *LatestFeedDataMap) GetLatestFeedDataFromCache(ctx context.Context, feedIds []int32) ([]*FeedData, error) { + if _, ok := ctx.Deadline(); !ok { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, 5*time.Second) + defer cancel() + }
92-95: Enhance error handling for cache operations.The error from db.MGetObject should be wrapped with more context about the operation.
- result, err := db.MGetObject[*FeedData](ctx, queryingKeys) - if err != nil { - return nil, err - } + result, err := db.MGetObject[*FeedData](ctx, queryingKeys) + if err != nil { + return nil, fmt.Errorf("failed to fetch feed data from cache: %w", err) + }
86-103: Consider performance implications with large feed ID lists.The method performs multiple operations that could be expensive with large feed ID lists:
- Slice allocation for keys
- Batch Redis operation
- In-memory state update
Consider adding a limit to the number of feed IDs that can be processed in a single call.
+const maxFeedIDsPerRequest = 1000 + func (m *LatestFeedDataMap) GetLatestFeedDataFromCache(ctx context.Context, feedIds []int32) ([]*FeedData, error) { + if len(feedIds) > maxFeedIDsPerRequest { + return nil, fmt.Errorf("too many feed IDs requested: %d (max: %d)", len(feedIds), maxFeedIDsPerRequest) + }node/pkg/fetcher/localaggregator.go (2)
236-241: Add error logging when falling back to cache.The fallback mechanism is well implemented, but it would be helpful to log the error when falling back to cache for better observability.
Consider adding logging:
result, err := c.latestFeedDataMap.GetLatestFeedData(feedIds) if err != nil || len(result) == 0 { + if err != nil { + log.Debug(). + Str("Player", "LocalAggregator"). + Err(err). + Msg("falling back to cache for feed data") + } return c.latestFeedDataMap.GetLatestFeedDataFromCache(ctx, feedIds) }
236-241: Consider adding metrics and timeout handling.The implementation would benefit from:
- Adding metrics to track cache fallbacks for monitoring
- Ensuring proper context timeout handling for the cache call
Consider:
- Adding Prometheus/metrics counter for cache fallbacks
- Using context with timeout for the cache call:
ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel()node/pkg/websocketfetcher/app.go (1)
308-312: Add metrics for cache operationsWhile the error handling is appropriate (log and continue), we should track cache failures to monitor system health. Consider adding metrics for:
- Cache operation success/failure rates
- Cache operation latency
- Batch sizes
This will help in:
- Early detection of Redis issues
- Performance optimization
- Capacity planning
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (4)
node/pkg/common/keys/keys.go(1 hunks)node/pkg/common/types/types.go(2 hunks)node/pkg/fetcher/localaggregator.go(1 hunks)node/pkg/websocketfetcher/app.go(3 hunks)
🔇 Additional comments (6)
node/pkg/common/keys/keys.go (2)
3-5: LGTM!
The strconv import is appropriately added for integer to string conversion in the new FeedData function.
11-13: Add documentation and input validation
While the implementation follows Redis key naming conventions and is consistent with existing patterns, consider these improvements:
- Add documentation explaining the key format and usage
- Add validation for negative feedID values
+// FeedData generates a Redis key for storing feed data.
+// The key format is "feedData:<feedID>" where feedID is a positive integer
+// that uniquely identifies a price feed.
func FeedData(feedID int32) string {
+ if feedID < 0 {
+ panic("feedID must be non-negative")
+ }
return "feedData:" + strconv.Itoa(int(feedID))
}Let's verify the key format consistency across the codebase:
node/pkg/common/types/types.go (2)
4-11: LGTM! Import changes are appropriate.
The new imports are necessary for the Redis caching functionality and follow Go conventions.
97-102: Potential race condition in cache update.
The SetLatestFeedData call could race with other goroutines updating the in-memory state. Consider adding a comment about this design choice or refactoring to prevent races.
Consider one of these approaches:
- Add explicit documentation about the race condition and why it's acceptable
- Use atomic operations or a more sophisticated synchronization mechanism
- Consider using Redis as the single source of truth instead of maintaining duplicate state
node/pkg/websocketfetcher/app.go (2)
318-325: Verify cache key format and race conditions
Please ensure:
- The cache key format matches the reader's expectations
- There are no race conditions between cache reads and direct reads
Let's verify the cache key format and usage:
✅ Verification successful
Cache implementation is consistent and well-structured
The verification reveals a well-implemented caching system:
-
Cache key format is consistent and properly structured:
- Keys are generated using
keys.FeedData(feedID)which creates keys in format "feedData:" - The same key format is used consistently across both write (
StoreIntoRedisCache) and read operations
- Keys are generated using
-
No race conditions are apparent between cache and direct reads:
- Cache writes use atomic
MSetObjectWithExpwith TTL - Cache reads use corresponding
MGetObjectoperations - The implementation includes proper error handling and type safety
- Cache writes use atomic
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for cache key usage
rg -l "FeedData\(" --type go | xargs ast-grep --pattern 'keys.FeedData($_)'
# Search for cache read operations
rg -l "GetLatestFeedDataFromCache" --type go | xargs ast-grep --pattern 'GetLatestFeedDataFromCache'
Length of output: 176
Script:
#!/bin/bash
# Let's try a broader search for cache-related operations and keys
# Search for key generation and cache operations
rg -A 5 "keys\.Feed" --type go
# Search for cache read/write operations
rg -A 5 "MSetObjectWithExp|GetObject" --type go
# Look for the keys package
fd -e go "keys\.go$" -x cat {}
Length of output: 4547
47-47: Document the rationale for the 15-second TTL
The warmCacheTTL of 15 seconds seems quite short and could lead to frequent cache misses if there are any WebSocket connection issues. Please document:
- Why 15 seconds was chosen
- What happens when the cache expires
- How this interacts with the WebSocket reconnection strategy
Let's verify the WebSocket reconnection timing:
✅ Verification successful
Let me search for more context about the WebSocket connection and caching strategy.
Based on the context gathered, I can now provide a final response:
15-second TTL is appropriate for real-time price feed caching
The warmCacheTTL of 15 seconds is well-justified for this use case because:
- It's used specifically for caching real-time price feed data from WebSocket connections
- The cache serves as a fallback mechanism during brief WebSocket interruptions
- Price data older than 15 seconds could be stale for trading purposes
- The WebSocket implementation has robust reconnection handling (every 100-500ms as seen in the tests)
The cache TTL aligns well with the WebSocket's reconnection strategy, providing a good balance between data freshness and connection resilience.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for reconnection logic and timeouts
rg -A 5 "reconnect|retry|timeout" --type go
Length of output: 20732
Script:
#!/bin/bash
# Search for WebSocket connection and caching related code in websocketfetcher
rg -A 10 "cache|websocket|feed" node/pkg/websocketfetcher/app.go
Length of output: 6039
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Outside diff range and nitpick comments (3)
node/pkg/websocketfetcher/app.go (1)
308-312: Optimize batch processing efficiencyThe current implementation processes one item before starting the batch loop. This could be optimized by collecting all items first, then processing them as a single batch.
Consider this optimization:
func (a *App) storeFeedData(ctx context.Context) { select { case <-ctx.Done(): return case feedData := <-a.buffer: batch := []*common.FeedData{feedData} - // Continue to drain the buffer until it's empty + // Collect all available items first loop: for { select { case feedData := <-a.buffer: batch = append(batch, feedData) a.feedDataDumpChannel <- feedData default: break loop } } + // Process the entire batch at once err := a.latestFeedDataMap.SetLatestFeedData(batch) if err != nil { log.Error().Err(err).Msg("error in setting latest feed data") } err = a.StoreIntoRedisCache(ctx, batch) if err != nil { log.Error().Err(err).Msg("error in storing into redis cache") } default: return } }node/pkg/common/types/types.go (2)
87-89: Consider returning an empty slice instead of nil whenfeedIdsis emptyReturning an empty slice can prevent potential
nilpointer dereferences in calling functions.Apply this diff to adjust the return value:
if len(feedIds) == 0 { - return nil, nil + return []*FeedData{}, nil }
101-106: Enhance error context when updating latest feed dataWrapping the error from
SetLatestFeedDatawith additional context improves debuggability.Consider modifying the error handling as follows:
if len(result) != 0 { err = m.SetLatestFeedData(result) if err != nil { - return nil, err + return nil, fmt.Errorf("failed to set latest feed data: %w", err) } }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (2)
node/pkg/common/types/types.go(2 hunks)node/pkg/websocketfetcher/app.go(3 hunks)
🔇 Additional comments (5)
node/pkg/websocketfetcher/app.go (1)
318-325: 🛠️ Refactor suggestion
Add input validation and monitoring metrics
The Redis caching implementation could benefit from:
- Input validation to prevent storing invalid data
- Monitoring metrics to track cache operations and performance
Consider these improvements:
func (a *App) StoreIntoRedisCache(ctx context.Context, batch []*types.FeedData) error {
+ // Input validation
+ for _, feedData := range batch {
+ if feedData == nil || feedData.FeedID == 0 {
+ return fmt.Errorf("invalid feed data detected: %+v", feedData)
+ }
+ }
+
+ // Start monitoring
+ start := time.Now()
+ defer func() {
+ metrics.CacheOperationDuration.WithLabelValues("store").Observe(time.Since(start).Seconds())
+ metrics.CachedItemsCount.WithLabelValues("feed_data").Add(float64(len(batch)))
+ }()
+
batchStoreEntries := make(map[string]any)
for _, feedData := range batch {
batchStoreEntries[keys.FeedData(feedData.FeedID)] = feedData
}
return db.MSetObjectWithExp(ctx, batchStoreEntries, warmCacheTTL)
}Run this script to check if metrics package is available:
node/pkg/common/types/types.go (4)
4-9: Appropriate imports of standard libraries
The imported packages are necessary for the new functionality introduced in the code.
10-11: Added imports are necessary for cache implementation
The addition of the "keys" and "db" packages is required for implementing the cache functionality.
92-94: Add input validation for feedIds
The method should validate each feedId to ensure they are positive integers, preventing unnecessary cache operations or errors due to invalid IDs.
86-107: Ensure thread safety when accessing FeedDataMap
Since SetLatestFeedData modifies FeedDataMap, verify that all accesses to FeedDataMap are properly synchronized to prevent data races.
Run the following script to check for unsynchronized access to FeedDataMap:
✅ Verification successful
Thread safety is properly implemented in LatestFeedDataMap
The code review concern has been thoroughly investigated and the implementation is correct. The LatestFeedDataMap struct has proper synchronization:
- It has a
sync.RWMutexfield namedMu - All accesses to
FeedDataMapare properly protected:- Read operations use
RLock()/RUnlock() - Write operations use
Lock()/Unlock()
- Read operations use
- No direct access to
FeedDataMapwas found outside of synchronized methods
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Search for accesses to FeedDataMap without proper locking mechanisms.
# Find all Go files except 'types.go'
fd --extension go --exclude 'types.go' --exec rg --fixed-strings 'FeedDataMap' {}
# Note: Review the findings to ensure that every access to 'FeedDataMap' is protected by the mutex.
Length of output: 1934
Script:
#!/bin/bash
# Let's examine the LatestFeedDataMap struct definition and its methods to check for mutex usage
ast-grep --pattern 'type LatestFeedDataMap struct {
$$$
}'
# Also search for any mutex-related fields and operations
rg -A 5 'sync\.(\w+)?Mutex' --type go
Length of output: 9529
Script:
#!/bin/bash
# Let's examine all methods of LatestFeedDataMap to verify mutex usage
ast-grep --pattern 'func (m *LatestFeedDataMap) $$$'
# Also check for direct access to FeedDataMap field
rg -B 2 -A 2 '\.FeedDataMap\[' --type go
Length of output: 841
| result, err := db.MGetObject[*FeedData](ctx, queryingKeys) | ||
| if err != nil { | ||
| return nil, err | ||
| } |
There was a problem hiding this comment.
Handle potential nil result from db.MGetObject
Ensure that the result from db.MGetObject is not nil before proceeding to prevent possible nil pointer dereferences.
Apply this diff to safeguard against nil results:
result, err := db.MGetObject[*FeedData](ctx, queryingKeys)
if err != nil {
return nil, err
}
+if result == nil {
+ result = []*FeedData{}
}📝 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.
| result, err := db.MGetObject[*FeedData](ctx, queryingKeys) | |
| if err != nil { | |
| return nil, err | |
| } | |
| result, err := db.MGetObject[*FeedData](ctx, queryingKeys) | |
| if err != nil { | |
| return nil, err | |
| } | |
| if result == nil { | |
| result = []*FeedData{} | |
| } |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Outside diff range and nitpick comments (1)
node/pkg/common/types/types.go (1)
86-111: Consider adding TTL and batch processing.A few suggestions to improve the cache implementation:
- Add TTL checks to prevent using stale data
- Add batch size limits for the Redis MGet operation to prevent memory issues
Example implementation:
const ( maxBatchSize = 100 cacheTTL = 5 * time.Minute ) func (m *LatestFeedDataMap) GetLatestFeedDataFromCache(ctx context.Context, feedIds []int32) ([]*FeedData, error) { if len(feedIds) == 0 { return nil, nil } // Process in batches var allResults []*FeedData for i := 0; i < len(feedIds); i += maxBatchSize { end := i + maxBatchSize if end > len(feedIds) { end = len(feedIds) } batch := feedIds[i:end] // Process batch... queryingKeys := make([]string, 0, len(batch)) for _, feedId := range batch { queryingKeys = append(queryingKeys, keys.FeedData(feedId)) } result, err := db.MGetObject[*FeedData](ctx, queryingKeys) if err != nil { return nil, err } // Filter out stale data now := time.Now() for _, data := range result { if data != nil && data.Timestamp != nil && now.Sub(*data.Timestamp) < cacheTTL { allResults = append(allResults, data) } } } // Update local cache asynchronously if len(allResults) > 0 { go func(data []*FeedData) { if err := m.SetLatestFeedData(data); err != nil { // Consider logging the error } }(allResults) } return allResults, nil }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (1)
node/pkg/common/types/types.go(2 hunks)
🔇 Additional comments (2)
node/pkg/common/types/types.go (2)
4-11: LGTM: Import changes are appropriate.
The new imports support the Redis caching functionality and are correctly organized.
86-111: Implementation meets PR objectives but needs reliability improvements.
The Redis caching implementation will help prevent errors during system restarts by providing access to latest prices. However, the identified issues around deadlocks, stale data, and batch processing should be addressed to ensure system stability.
Let's verify the Redis implementation:
✅ Verification successful
Let me gather more information about the Redis implementation, particularly focusing on error handling and connection management in the codebase.
Let me gather more information about the Redis implementation, particularly focusing on the caching mechanism and error handling in the FeedData related code.
Based on the gathered information, I can now provide a final assessment of the Redis implementation. The code shows proper Redis connection management, error handling, and caching mechanisms:
Redis implementation is reliable and meets PR objectives
- Proper connection management with deferred closing:
defer redisConnection.Close() - Robust error handling in Redis operations with appropriate logging
- Thread-safe cache operations using mutex locks in
SetLatestFeedData - Efficient batch operations using
MGetObjectfor retrieving multiple feed data - Graceful handling of empty results and cache misses
- Appropriate error propagation to calling functions
The implementation successfully addresses the PR's goal of preventing errors during system restarts by caching latest prices, with proper reliability considerations built in.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Check if Redis-related configurations and error handling are properly set up
# Look for Redis configuration
rg -A 5 'redis\.(New|Connect|Config)'
# Check for Redis error handling patterns
rg -A 3 'redis.*err\s*:?='
# Look for potential Redis connection leaks
ast-grep --pattern 'redis.New$$$'
Length of output: 8511
Script:
#!/bin/bash
# Check Redis-related error handling and connection management
# Look for Redis connection management patterns
rg -A 5 'defer.*redis.*Close'
# Check Redis operations in the FeedData related code
rg -A 5 'MGetObject.*FeedData'
# Look for Redis connection pool configurations
rg 'PoolSize|MaxRetries|MinIdleConns'
# Check Redis timeout configurations
rg 'Timeout|ReadTimeout|WriteTimeout|DialTimeout'
Length of output: 136133
Script:
#!/bin/bash
# Check FeedData related Redis implementation
# Look for FeedData struct and its usage
rg -A 10 'type FeedData struct'
# Check Redis cache operations in FeedData
rg -A 5 'SetLatestFeedData|GetLatestFeedDataFromCache'
# Look for Redis error handling in db package
rg -A 5 'func.*MGetObject'
# Check Redis connection management in db package
rg -A 10 'func.*NewRedis'
Length of output: 5190
| if len(result) != 0 { | ||
| err = m.SetLatestFeedData(result) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| } |
There was a problem hiding this comment.
Potential deadlock risk in cache update.
The SetLatestFeedData method acquires a write lock while GetLatestFeedDataFromCache might be called concurrently. Consider updating the local cache asynchronously or ensuring proper lock ordering to prevent potential deadlocks.
Suggested fix:
if len(result) != 0 {
- err = m.SetLatestFeedData(result)
- if err != nil {
- return nil, err
- }
+ go func(data []*FeedData) {
+ if err := m.SetLatestFeedData(data); err != nil {
+ // Consider logging the error
+ }
+ }(result)
}Committable suggestion skipped: line range outside the PR's diff.
Description
local aggregator fails to load latest price data until the initial price has been loaded through fetcher. this update will store latest prices from fetcher into redis so that it could avoid such error on restart
Type of change
Please delete options that are not relevant.
Checklist before requesting a review
Deployment