oib/aitbc
1
0
Fork 0
Code Issues Pull Requests Actions 7 Packages Projects Releases Wiki Activity

refactor(coordinator-api,blockchain-explorer): add response caching and fix timestamp handling

Browse Source 
- Add cached decorator to admin stats, job status, payment status, and marketplace stats endpoints
- Configure cache TTLs using get_cache_config for different endpoint types (1min job_list, 30s user_balance, marketplace_stats)
- Import cache_management router and include it in main app with /v1 prefix
- Fix blockchain-explorer formatTimestamp to handle both ISO string and Unix numeric timestamps with type
This commit is contained in:
oib
2026-02-28 21:50:25 +01:00
parent 93ffaf53de
commit 2d97783fb1
13 changed files with 1030 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
apps/blockchain-explorer/main.py
View File

@@ -362,10 +362,29 @@ HTML_TEMPLATE = r"""
alert('Search by block height or transaction hash (64 char hex) is supported');
}
// Format timestamp
// Format timestamp - robust for both numeric and ISO string timestamps
function formatTimestamp(timestamp) {
if (!timestamp) return '-';
return new Date(timestamp * 1000).toLocaleString();
// Handle ISO string timestamps
if (typeof timestamp === 'string') {
try {
return new Date(timestamp).toLocaleString();
} catch (e) {
return '-';
}
}
// Handle numeric timestamps (Unix seconds)
if (typeof timestamp === 'number') {
try {
return new Date(timestamp * 1000).toLocaleString();
} catch (e) {
return '-';
}
}
return '-';
}
// Auto-refresh every 30 seconds
@@ -376,15 +395,15 @@ HTML_TEMPLATE = r"""
"""
async def get_chain_head() -> Dict[str, Any]:
"""Get the current chain head"""
async def get_transaction(tx_hash: str) -> Dict[str, Any]:
"""Get transaction by hash"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/head")
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/tx/{tx_hash}")
if response.status_code == 200:
return response.json()
except Exception as e:
print(f"Error getting chain head: {e}")
print(f"Error getting transaction: {e}")
return {}
@@ -424,7 +443,7 @@ async def api_transaction(tx_hash: str):
"""API endpoint for transaction data, normalized for frontend"""
async with httpx.AsyncClient() as client:
try:
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/tx/{tx_hash}")
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/tx/{tx_hash}")
if response.status_code == 200:
tx = response.json()
# Normalize for frontend expectations

4
apps/coordinator-api/src/app/main.py
View File

@@ -25,7 +25,8 @@ from .routers import (
explorer,
payments,
web_vitals,
edge_gpu
edge_gpu,
cache_management
)
from .routers.ml_zk_proofs import router as ml_zk_proofs
from .routers.community import router as community_router
@@ -221,6 +222,7 @@ def create_app() -> FastAPI:
app.include_router(openclaw_enhanced, prefix="/v1")
app.include_router(monitoring_dashboard, prefix="/v1")
app.include_router(multi_modal_rl_router, prefix="/v1")
app.include_router(cache_management, prefix="/v1")
# Add Prometheus metrics endpoint
metrics_app = make_asgi_app()

2
apps/coordinator-api/src/app/routers/__init__.py
View File

@@ -13,6 +13,7 @@ from .marketplace_offers import router as marketplace_offers
from .payments import router as payments
from .web_vitals import router as web_vitals
from .edge_gpu import router as edge_gpu
from .cache_management import router as cache_management
# from .registry import router as registry
__all__ = [
@@ -29,5 +30,6 @@ __all__ = [
"payments",
"web_vitals",
"edge_gpu",
"cache_management",
"registry",
]

5
apps/coordinator-api/src/app/routers/admin.py
View File

@@ -6,6 +6,8 @@ from slowapi.util import get_remote_address
from ..deps import require_admin_key
from ..services import JobService, MinerService
from ..storage import SessionDep
from ..utils.cache import cached, get_cache_config
from ..config import settings
from aitbc.logging import get_logger
logger = get_logger(__name__)
@@ -14,7 +16,8 @@ router = APIRouter(prefix="/admin", tags=["admin"])
@router.get("/stats", summary="Get coordinator stats")
@limiter.limit("20/minute")
@limiter.limit(lambda: settings.rate_limit_admin_stats)
@cached(**get_cache_config("job_list")) # Cache admin stats for 1 minute
async def get_stats(
request: Request,
session: SessionDep,

111
apps/coordinator-api/src/app/routers/cache_management.py Normal file
View File

@@ -0,0 +1,111 @@
"""
Cache monitoring and management endpoints
"""
from fastapi import APIRouter, Depends, HTTPException, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from ..deps import require_admin_key
from ..utils.cache_management import get_cache_stats, clear_cache, warm_cache
from ..config import settings
from aitbc.logging import get_logger
logger = get_logger(__name__)
limiter = Limiter(key_func=get_remote_address)
router = APIRouter(prefix="/cache", tags=["cache-management"])
@router.get("/stats", summary="Get cache statistics")
@limiter.limit(lambda: settings.rate_limit_admin_stats)
async def get_cache_statistics(
request: Request,
admin_key: str = Depends(require_admin_key())
):
"""Get cache performance statistics"""
try:
stats = get_cache_stats()
return {
"cache_health": stats,
"status": "healthy" if stats["health_status"] in ["excellent", "good"] else "degraded"
}
except Exception as e:
logger.error(f"Failed to get cache stats: {e}")
raise HTTPException(status_code=500, detail="Failed to retrieve cache statistics")
@router.post("/clear", summary="Clear cache entries")
@limiter.limit(lambda: settings.rate_limit_admin_stats)
async def clear_cache_entries(
request: Request,
pattern: str = None,
admin_key: str = Depends(require_admin_key())
):
"""Clear cache entries (all or matching pattern)"""
try:
result = clear_cache(pattern)
logger.info(f"Cache cleared by admin: pattern={pattern}, result={result}")
return result
except Exception as e:
logger.error(f"Failed to clear cache: {e}")
raise HTTPException(status_code=500, detail="Failed to clear cache")
@router.post("/warm", summary="Warm up cache")
@limiter.limit(lambda: settings.rate_limit_admin_stats)
async def warm_up_cache(
request: Request,
admin_key: str = Depends(require_admin_key())
):
"""Trigger cache warming for common queries"""
try:
result = warm_cache()
logger.info("Cache warming triggered by admin")
return result
except Exception as e:
logger.error(f"Failed to warm cache: {e}")
raise HTTPException(status_code=500, detail="Failed to warm cache")
@router.get("/health", summary="Get cache health status")
@limiter.limit(lambda: settings.rate_limit_admin_stats)
async def cache_health_check(
request: Request,
admin_key: str = Depends(require_admin_key())
):
"""Get detailed cache health information"""
try:
from ..utils.cache import cache_manager
stats = get_cache_stats()
cache_data = cache_manager.get_stats()
return {
"health": stats,
"detailed_stats": cache_data,
"recommendations": _get_cache_recommendations(stats)
}
except Exception as e:
logger.error(f"Failed to get cache health: {e}")
raise HTTPException(status_code=500, detail="Failed to retrieve cache health")
def _get_cache_recommendations(stats: dict) -> list:
"""Get cache performance recommendations"""
recommendations = []
hit_rate = stats["hit_rate_percent"]
total_entries = stats["total_entries"]
if hit_rate < 40:
recommendations.append("Low hit rate detected. Consider increasing cache TTL or warming cache more frequently.")
if total_entries > 10000:
recommendations.append("High number of cache entries. Consider implementing cache size limits or more aggressive cleanup.")
if hit_rate > 95:
recommendations.append("Very high hit rate. Cache TTL might be too long, consider reducing for fresher data.")
if not recommendations:
recommendations.append("Cache performance is optimal.")
return recommendations

2
apps/coordinator-api/src/app/routers/client.py
View File

@@ -9,6 +9,7 @@ from ..services import JobService
from ..services.payments import PaymentService
from ..config import settings
from ..storage import SessionDep
from ..utils.cache import cached, get_cache_config
limiter = Limiter(key_func=get_remote_address)
router = APIRouter(tags=["client"])
@@ -44,6 +45,7 @@ async def submit_job(
@router.get("/jobs/{job_id}", response_model=JobView, summary="Get job status")
@cached(**get_cache_config("job_list")) # Cache job status for 1 minute
async def get_job(
job_id: str,
session: SessionDep,

3
apps/coordinator-api/src/app/routers/exchange.py
View File

@@ -25,6 +25,8 @@ from ..schemas import (
WalletInfoResponse
)
from ..services.bitcoin_wallet import get_wallet_balance, get_wallet_info
from ..utils.cache import cached, get_cache_config
from ..config import settings
router = APIRouter(tags=["exchange"])
@@ -85,6 +87,7 @@ async def create_payment(
@router.get("/exchange/payment-status/{payment_id}", response_model=PaymentStatusResponse)
@cached(**get_cache_config("user_balance")) # Cache payment status for 30 seconds
async def get_payment_status(payment_id: str) -> Dict[str, Any]:
"""Get payment status"""

5
apps/coordinator-api/src/app/routers/marketplace.py
View File

@@ -9,6 +9,8 @@ from ..schemas import MarketplaceBidRequest, MarketplaceOfferView, MarketplaceSt
from ..services import MarketplaceService
from ..storage import SessionDep
from ..metrics import marketplace_requests_total, marketplace_errors_total
from ..utils.cache import cached, get_cache_config
from ..config import settings
from aitbc.logging import get_logger
logger = get_logger(__name__)
@@ -51,7 +53,8 @@ async def list_marketplace_offers(
response_model=MarketplaceStatsView,
summary="Get marketplace summary statistics",
)
@limiter.limit("50/minute")
@limiter.limit(lambda: settings.rate_limit_marketplace_stats)
@cached(**get_cache_config("marketplace_stats"))
async def get_marketplace_stats(
request: Request,
*,

237
apps/coordinator-api/src/app/utils/cache_management.py Normal file
View File

@@ -0,0 +1,237 @@
"""
Cache management utilities for endpoints
"""
from ..utils.cache import cache_manager, cleanup_expired_cache
from ..config import settings
from aitbc.logging import get_logger
logger = get_logger(__name__)
def invalidate_cache_pattern(pattern: str):
"""Invalidate cache entries matching a pattern"""
keys_to_delete = []
for key in cache_manager._cache.keys():
if pattern in key:
keys_to_delete.append(key)
for key in keys_to_delete:
cache_manager.delete(key)
logger.info(f"Invalidated {len(keys_to_delete)} cache entries matching pattern: {pattern}")
return len(keys_to_delete)
def get_cache_health() -> dict:
"""Get cache health statistics"""
stats = cache_manager.get_stats()
# Determine health status
total_requests = stats["total_requests"]
if total_requests == 0:
hit_rate = 0
health_status = "unknown"
else:
hit_rate = stats["hit_rate_percent"]
if hit_rate >= 80:
health_status = "excellent"
elif hit_rate >= 60:
health_status = "good"
elif hit_rate >= 40:
health_status = "fair"
else:
health_status = "poor"
return {
"health_status": health_status,
"hit_rate_percent": hit_rate,
"total_entries": stats["total_entries"],
"total_requests": total_requests,
"memory_usage_mb": round(len(str(cache_manager._cache)) / 1024 / 1024, 2),
"last_cleanup": stats.get("last_cleanup", "never")
}
# Cache invalidation strategies for different events
class CacheInvalidationStrategy:
"""Strategies for cache invalidation based on events"""
@staticmethod
def on_job_created(job_id: str):
"""Invalidate caches when a job is created"""
# Invalidate job list caches
invalidate_cache_pattern("jobs_")
invalidate_cache_pattern("admin_stats")
logger.info(f"Invalidated job-related caches for new job: {job_id}")
@staticmethod
def on_job_updated(job_id: str):
"""Invalidate caches when a job is updated"""
# Invalidate specific job cache and lists
invalidate_cache_pattern(f"jobs_get_job_{job_id}")
invalidate_cache_pattern("jobs_")
invalidate_cache_pattern("admin_stats")
logger.info(f"Invalidated job caches for updated job: {job_id}")
@staticmethod
def on_marketplace_change():
"""Invalidate caches when marketplace data changes"""
invalidate_cache_pattern("marketplace_")
logger.info("Invalidated marketplace caches due to data change")
@staticmethod
def on_payment_created(payment_id: str):
"""Invalidate caches when a payment is created"""
invalidate_cache_pattern("balance_")
invalidate_cache_pattern("payment_")
invalidate_cache_pattern("admin_stats")
logger.info(f"Invalidated payment caches for new payment: {payment_id}")
@staticmethod
def on_payment_updated(payment_id: str):
"""Invalidate caches when a payment is updated"""
invalidate_cache_pattern(f"balance_")
invalidate_cache_pattern(f"payment_{payment_id}")
logger.info(f"Invalidated payment caches for updated payment: {payment_id}")
# Background task for cache management
async def cache_management_task():
"""Background task for cache maintenance"""
while True:
try:
# Clean up expired entries
removed_count = cleanup_expired_cache()
# Log cache health periodically
if removed_count > 0:
health = get_cache_health()
logger.info(f"Cache cleanup completed: {removed_count} entries removed, "
f"hit rate: {health['hit_rate_percent']}%, "
f"entries: {health['total_entries']}")
# Run cache management every 5 minutes
import asyncio
await asyncio.sleep(300)
except Exception as e:
logger.error(f"Cache management error: {e}")
await asyncio.sleep(60) # Retry after 1 minute on error
# Cache warming utilities for startup
class CacheWarmer:
"""Cache warming utilities for common endpoints"""
def __init__(self, session):
self.session = session
async def warm_common_queries(self):
"""Warm up cache with common queries"""
try:
logger.info("Starting cache warming...")
# Warm marketplace stats (most commonly accessed)
await self._warm_marketplace_stats()
# Warm admin stats
await self._warm_admin_stats()
# Warm exchange rates
await self._warm_exchange_rates()
logger.info("Cache warming completed successfully")
except Exception as e:
logger.error(f"Cache warming failed: {e}")
async def _warm_marketplace_stats(self):
"""Warm marketplace statistics cache"""
try:
from ..services.marketplace import MarketplaceService
service = MarketplaceService(self.session)
stats = service.get_stats()
# Manually cache the result
from ..utils.cache import cache_manager
cache_manager.set("marketplace_stats_get_marketplace_stats", stats, ttl_seconds=300)
logger.info("Marketplace stats cache warmed")
except Exception as e:
logger.warning(f"Failed to warm marketplace stats: {e}")
async def _warm_admin_stats(self):
"""Warm admin statistics cache"""
try:
from ..services import JobService, MinerService
from sqlmodel import func, select
from ..domain import Job
job_service = JobService(self.session)
miner_service = MinerService(self.session)
# Simulate admin stats query
total_jobs = self.session.exec(select(func.count()).select_from(Job)).one()
active_jobs = self.session.exec(select(func.count()).select_from(Job).where(Job.state.in_(["QUEUED", "RUNNING"]))).one()
miners = miner_service.list_records()
stats = {
"total_jobs": int(total_jobs or 0),
"active_jobs": int(active_jobs or 0),
"online_miners": miner_service.online_count(),
"avg_miner_job_duration_ms": 0,
}
# Manually cache the result
from ..utils.cache import cache_manager
cache_manager.set("job_list_get_stats", stats, ttl_seconds=60)
logger.info("Admin stats cache warmed")
except Exception as e:
logger.warning(f"Failed to warm admin stats: {e}")
async def _warm_exchange_rates(self):
"""Warm exchange rates cache"""
try:
# Mock exchange rates - in production this would call an exchange API
rates = {
"AITBC_BTC": 0.00001,
"AITBC_USD": 0.10,
"BTC_USD": 50000.0
}
# Manually cache the result
from ..utils.cache import cache_manager
cache_manager.set("rates_current", rates, ttl_seconds=600)
logger.info("Exchange rates cache warmed")
except Exception as e:
logger.warning(f"Failed to warm exchange rates: {e}")
# FastAPI endpoints for cache management
async def get_cache_stats():
"""Get cache statistics (for monitoring)"""
return get_cache_health()
async def clear_cache(pattern: str = None):
"""Clear cache entries"""
if pattern:
count = invalidate_cache_pattern(pattern)
return {"status": "cleared", "pattern": pattern, "count": count}
else:
cache_manager.clear()
return {"status": "cleared", "pattern": "all", "count": "all"}
async def warm_cache():
"""Manually trigger cache warming"""
# This would need to be called with a session
# For now, just return status
return {"status": "cache_warming_triggered"}