feat: snapshotter active status check at runtime

snapshotter/core_api.py

@@ -36,6 +36,7 @@
 from snapshotter.utils.redis.redis_keys import epoch_id_project_to_state_mapping
 from snapshotter.utils.redis.redis_keys import epoch_process_report_cached_key
 from snapshotter.utils.redis.redis_keys import project_last_finalized_epoch_key
+from snapshotter.utils.redis.redis_keys import active_status_key
 from snapshotter.utils.rpc import RpcHelper
 
 
@@ -97,11 +98,23 @@ async def startup_boilerplate():
     await app.state.ipfs_singleton.init_sessions()
     app.state.ipfs_reader_client = app.state.ipfs_singleton._ipfs_read_client
 
-# Health check endpoint that returns 200 OK
-
 
+# Health check endpoint
 @app.get('/health')
-async def health_check():
+async def health_check(
+    request: Request,
+    response: Response,
+):
+    redis_conn: aioredis.Redis = request.app.state.redis_pool
+    _ = await redis_conn.get(active_status_key)
+    if _:
+        active_status = bool(int(_))
+        if not active_status:
+            response.status_code = 503
+            return {
+                'status': 'error',
+                'message': 'Snapshotter is not active',
+            }
     return {'status': 'OK'}
 
 # get current epoch

snapshotter/processor_distributor.py

@@ -41,6 +41,7 @@
 from snapshotter.utils.models.data_models import PreloaderAsyncFutureDetails
 from snapshotter.utils.models.data_models import SnapshotterStates
 from snapshotter.utils.models.data_models import SnapshotterStateUpdate
+from snapshotter.utils.models.data_models import SnapshottersUpdatedEvent
 from snapshotter.utils.models.message_models import EpochBase
 from snapshotter.utils.models.message_models import PayloadCommitFinalizedMessage
 from snapshotter.utils.models.message_models import PowerloomCalculateAggregateMessage
@@ -54,6 +55,7 @@
 from snapshotter.utils.redis.redis_keys import epoch_id_project_to_state_mapping
 from snapshotter.utils.redis.redis_keys import project_finalized_data_zset
 from snapshotter.utils.redis.redis_keys import snapshot_submission_window_key
+from snapshotter.utils.redis.redis_keys import active_status_key
 from snapshotter.utils.rpc import RpcHelper
 
 
@@ -674,7 +676,14 @@ async def _on_rabbitmq_message(self, message: IncomingMessage):
             )
         elif message_type == 'ProjectsUpdated':
             await self._update_all_projects(message)
-
+        elif message_type == 'SnapshottersUpdated':
+            msg_cast = SnapshottersUpdatedEvent.parse_raw(message.body)
+            if msg_cast.snapshotterAddress == settings.instance_id:
+                if self._redis_conn:
+                    await self._redis_conn.set(
+                        active_status_key,
+                        int(msg_cast.allowed)
+                    )
         else:
             self._logger.error(
                 (

snapshotter/snapshotter_id_ping.py

@@ -4,6 +4,7 @@
 from snapshotter.auth.helpers.redis_conn import RedisPoolCache
 from snapshotter.settings.config import settings
 from snapshotter.utils.file_utils import read_json_file
+from snapshotter.utils.redis.redis_keys import active_status_key
 from snapshotter.utils.rpc import RpcHelper
 
 
@@ -28,6 +29,10 @@ async def main():
     allowed_snapshotters = snapshotters_arr_query[0]
     if settings.instance_id in allowed_snapshotters:
         print('Snapshotting allowed...')
+        await redis_conn.set(
+            active_status_key,
+            int(True)
+        )
         sys.exit(0)
     else:
         print('Snapshotting not allowed...')

snapshotter/system_event_detector.py

@@ -21,6 +21,7 @@
 from snapshotter.utils.models.data_models import EventBase
 from snapshotter.utils.models.data_models import ProjectsUpdatedEvent
 from snapshotter.utils.models.data_models import SnapshotFinalizedEvent
+from snapshotter.utils.models.data_models import SnapshottersUpdatedEvent
 from snapshotter.utils.rabbitmq_helpers import RabbitmqThreadedSelectLoopInteractor
 from snapshotter.utils.redis.redis_conn import RedisPoolCache
 from snapshotter.utils.redis.redis_keys import event_detector_last_processed_block
@@ -119,12 +120,15 @@ def __init__(self, name, **kwargs):
             'EpochReleased': self.contract.events.EpochReleased._get_event_abi(),
             'SnapshotFinalized': self.contract.events.SnapshotFinalized._get_event_abi(),
             'ProjectsUpdated': self.contract.events.ProjectsUpdated._get_event_abi(),
+            'SnapshottersUpdated': self.contract.events.SnapshottersUpdated._get_event_abi(),
         }
 
         EVENT_SIGS = {
             'EpochReleased': 'EpochReleased(uint256,uint256,uint256,uint256)',
             'SnapshotFinalized': 'SnapshotFinalized(uint256,uint256,string,string,uint256)',
             'ProjectsUpdated': 'ProjectsUpdated(string,bool,uint256)',
+            'SnapshottersUpdated': 'SnapshottersUpdated(address,bool)',
+
         }
 
         self.event_sig, self.event_abi = get_event_sig_and_abi(
@@ -187,6 +191,13 @@ async def get_events(self, from_block: int, to_block: int):
                     timestamp=int(time.time()),
                 )
                 events.append((log.event, event))
+            elif log.event == 'SnapshottersUpdated':
+                event = SnapshottersUpdatedEvent(
+                    snapshotterAddress=log.args.snapshotterAddress,
+                    allowed=log.args.allowed,
+                    timestamp=int(time.time()),
+                )
+                events.append((log.event, event))
 
         self._logger.info('Events: {}', events)
         return events

snapshotter/utils/models/data_models.py

@@ -108,6 +108,11 @@ class ProjectsUpdatedEvent(EventBase):
     enableEpochId: int
 
 
+class SnapshottersUpdatedEvent(EventBase):
+    snapshotterAddress: str
+    allowed: bool
+
+
 class SnapshotSubmittedEvent(EventBase):
     snapshotCid: str
     epochId: int

snapshotter/utils/redis/redis_keys.py

@@ -36,6 +36,8 @@
 
 snapshot_submission_window_key = 'snapshotSubmissionWindow'
 
+active_status_key = f'snapshotterActiveStatus:{settings.namespace}'
+
 # project finalzed data zset