97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
from typing import Dict, Any, Optional
|
|
import time
|
|
|
|
|
|
class CachePolicy:
|
|
"""
|
|
Determines caching behavior for Solana RPC methods based on their characteristics.
|
|
"""
|
|
|
|
# Methods that return immutable data - cache indefinitely
|
|
CACHEABLE_IMMUTABLE = {
|
|
'getGenesisHash' # Network genesis hash never changes
|
|
}
|
|
|
|
# Methods with time-based TTL caching based on data change frequency
|
|
CACHEABLE_WITH_TTL = {
|
|
# Network/validator information - changes periodically
|
|
'getVoteAccounts': 120, # Validator vote accounts change every few minutes
|
|
'getSupply': 300, # Total SOL supply changes slowly
|
|
|
|
# Epoch and network info - changes with epoch boundaries (~2-3 days)
|
|
'getEpochInfo': 3600, # Current epoch info changes slowly
|
|
'getInflationRate': 1800, # Inflation rate changes infrequently
|
|
'getInflationGovernor': 3600, # Inflation governor params rarely change
|
|
|
|
# Network constants - change very rarely or never
|
|
'getEpochSchedule': 86400, # Epoch schedule rarely changes
|
|
'getVersion': 3600, # RPC version changes occasionally
|
|
'getIdentity': 3600, # Node identity changes rarely
|
|
|
|
# Never change for the given parameters but will add new entry in the DB if the input parameters change
|
|
'getBlock': 86400,
|
|
'getTransaction':86400
|
|
}
|
|
|
|
def should_cache(self, method: str, params: Dict[str, Any]) -> bool:
|
|
"""
|
|
Determine if a method should be cached based on the method name and parameters.
|
|
|
|
Args:
|
|
method: The RPC method name
|
|
params: The method parameters
|
|
|
|
Returns:
|
|
True if the method should be cached, False otherwise
|
|
"""
|
|
|
|
if method in self.CACHEABLE_WITH_TTL:
|
|
# For getBlock, only cache finalized blocks
|
|
if method == 'getBlock':
|
|
commitment = self._get_commitment(params)
|
|
return commitment == 'finalized'
|
|
return True
|
|
|
|
if method in self.CACHEABLE_IMMUTABLE:
|
|
return True
|
|
|
|
# Default to not caching unknown methods
|
|
return False
|
|
|
|
def get_cache_ttl(self, method: str, params: Dict[str, Any]) -> Optional[int]:
|
|
"""
|
|
Get the Time To Live (TTL) for a cached method in seconds.
|
|
|
|
Args:
|
|
method: The RPC method name
|
|
params: The method parameters
|
|
|
|
Returns:
|
|
TTL in seconds, or None for indefinite caching
|
|
"""
|
|
if method in self.CACHEABLE_IMMUTABLE:
|
|
return None # Cache indefinitely
|
|
|
|
if method in self.CACHEABLE_WITH_TTL:
|
|
return self.CACHEABLE_WITH_TTL[method]
|
|
|
|
return None
|
|
|
|
def _get_commitment(self, params: Dict[str, Any]) -> str:
|
|
"""
|
|
Extract the commitment level from RPC parameters.
|
|
|
|
Args:
|
|
params: The method parameters
|
|
|
|
Returns:
|
|
The commitment level, defaults to 'processed'
|
|
"""
|
|
if isinstance(params, list) and len(params) > 1:
|
|
if isinstance(params[1], dict) and 'commitment' in params[1]:
|
|
return params[1]['commitment']
|
|
elif isinstance(params, dict) and 'commitment' in params:
|
|
return params['commitment']
|
|
|
|
return 'processed' # Default commitment level
|