1+ import json
12from fastapi import FastAPI
23from fastapi import Request
34from fastapi import Response
78from ipfs_client .main import AsyncIPFSClientSingleton
89from pydantic import Field
910from web3 import Web3
11+ import asyncio
12+ import time
13+ from pathlib import Path
14+ from httpx import AsyncClient , Limits , Timeout , AsyncHTTPTransport
1015
1116from snapshotter .settings .config import settings
17+ from snapshotter .utils .callback_helpers import send_telegram_notification_async
1218from snapshotter .utils .data_utils import get_project_epoch_snapshot
1319from snapshotter .utils .data_utils import get_project_finalized_cid
1420from snapshotter .utils .default_logger import logger
1521from snapshotter .utils .file_utils import read_json_file
16- from snapshotter .utils .models .data_models import TaskStatusRequest
22+ from snapshotter .utils .models .data_models import SnapshotterIssue , SnapshotterReportState , SnapshotterStatus , TaskStatusRequest
23+ from snapshotter .utils .models .message_models import TelegramSnapshotterReportMessage
1724from snapshotter .utils .rpc import RpcHelper
1825
1926
4451)
4552
4653
54+ async def check_last_submission ():
55+ while True :
56+ try :
57+ submission_file = Path ('last_successful_submission.txt' )
58+ if submission_file .exists ():
59+ last_timestamp = int (submission_file .read_text ().strip ())
60+ current_time = int (time .time ())
61+
62+ # If more than 5 minutes have passed since last submission
63+ if current_time - last_timestamp > 300 :
64+ rest_logger .error (
65+ 'No successful submission in the last 5 minutes. Last submission: {}' ,
66+ time .strftime ('%Y-%m-%d %H:%M:%S' , time .localtime (last_timestamp ))
67+ )
68+ # Send Telegram notification
69+ if settings .reporting .telegram_url and settings .reporting .telegram_chat_id :
70+ notification_message = SnapshotterIssue (
71+ instanceID = settings .instance_id ,
72+ issueType = SnapshotterReportState .UNHEALTHY_EPOCH_PROCESSING .value ,
73+ projectID = '' ,
74+ epochId = '' ,
75+ timeOfReporting = str (time .time ()),
76+ extra = json .dumps ({
77+ 'issueDetails' : f'No successful submission in the last 5 minutes. Last submission: { time .strftime ("%Y-%m-%d %H:%M:%S" , time .localtime (last_timestamp ))}
78+ }),
79+ )
80+
81+ telegram_message = TelegramSnapshotterReportMessage (
82+ chatId = settings .reporting .telegram_chat_id ,
83+ slotId = settings .slot_id ,
84+ issue = notification_message ,
85+ status = SnapshotterStatus (
86+ projects = [],
87+ totalMissedSubmissions = 0 ,
88+ consecutiveMissedSubmissions = 0 ,
89+ ),
90+ )
91+
92+ await send_telegram_notification_async (
93+ client = app .state .telegram_client ,
94+ message = telegram_message ,
95+ )
96+ app .state .healthy = False
97+ await asyncio .sleep (10 ) # Check every 10 seconds
98+
99+ except Exception as e :
100+ rest_logger .error ('Error checking last submission: {}' , e )
101+ await asyncio .sleep (10 ) # Still wait before retrying
102+
103+
47104@app .on_event ('startup' )
48105async def startup_boilerplate ():
49106 """
@@ -59,13 +116,32 @@ async def startup_boilerplate():
59116 abi = protocol_state_contract_abi ,
60117 )
61118
119+ # Initialize httpx client for Telegram notifications
120+ transport_limits = Limits (
121+ max_connections = 10 ,
122+ max_keepalive_connections = 5 ,
123+ keepalive_expiry = None ,
124+ )
125+
126+ app .state .telegram_client = AsyncClient (
127+ base_url = settings .reporting .telegram_url ,
128+ timeout = Timeout (timeout = 5.0 ),
129+ follow_redirects = False ,
130+ transport = AsyncHTTPTransport (limits = transport_limits ),
131+ )
132+
62133 if not settings .ipfs .url :
63134 rest_logger .warning ('IPFS url not set, /data API endpoint will be unusable!' )
64135 else :
65136 app .state .ipfs_singleton = AsyncIPFSClientSingleton (settings .ipfs )
66137 await app .state .ipfs_singleton .init_sessions ()
67138 app .state .ipfs_reader_client = app .state .ipfs_singleton ._ipfs_read_client
68139 app .state .epoch_size = 0
140+ app .state .healthy = True
141+ # Start the background task
142+ app .state .background_tasks = []
143+ background_task = asyncio .create_task (check_last_submission ())
144+ app .state .background_tasks .append (background_task )
69145
70146
71147# Health check endpoint
@@ -84,7 +160,11 @@ async def health_check(
84160 Returns:
85161 dict: A dictionary containing the status of the service.
86162 """
87- return {'status' : 'OK' }
163+ if app .state .healthy :
164+ return {'status' : 'OK' }
165+ else :
166+ response .status_code = 500
167+ return {'status' : 'UNHEALTHY' }
88168
89169
90170@app .get ('/current_epoch' )
@@ -259,6 +339,7 @@ async def get_data_for_project_id_epoch_id(
259339 'status' : 'error' ,
260340 'message' : f'IPFS url not set, /data API endpoint is unusable, please use /cid endpoint instead!' ,
261341 }
342+ # FIXME: outdated method signature
262343 try :
263344 data = await get_project_epoch_snapshot (
264345 request .app .state .protocol_state_contract ,
@@ -397,3 +478,14 @@ async def get_task_status_post(
397478 'completed' : False ,
398479 'message' : f'Task { task_status_request .task_type } { task_status_request .wallet_address } ,
399480 }
481+
482+
483+ @app .on_event ("shutdown" )
484+ async def shutdown_event ():
485+ """Cleanup background tasks"""
486+ for task in app .state .background_tasks :
487+ task .cancel ()
488+ try :
489+ await task
490+ except asyncio .CancelledError :
491+ pass
0 commit comments