Merge pull request #432 from ch4r10t33r/pr-logging

Add frequent node logging: attestations, blocks, justified/finalized, peers

Author: ch4r10t33r

src/lean_spec/subspecs/networking/service/service.py

@@ -144,17 +144,17 @@ async def _handle_event(self, event: NetworkEvent) -> None:
                 )
                 await self.sync_service.on_gossip_block(block, peer_id)
 
-            case GossipAttestationEvent(attestation=attestation):
-                #
-                # SyncService will validate signature and update forkchoice.
-                await self.sync_service.on_gossip_attestation(attestation)
+            case GossipAttestationEvent(attestation=attestation, peer_id=peer_id):
+                # SyncService validates signature and updates forkchoice.
+                # Logs attestation source and validation result.
+                await self.sync_service.on_gossip_attestation(attestation, peer_id)
 
-            case GossipAggregatedAttestationEvent(signed_attestation=att):
+            case GossipAggregatedAttestationEvent(signed_attestation=att, peer_id=peer_id):
                 # Route aggregated attestations to sync service.
                 #
                 # Aggregates contain multiple validator votes and are used
                 # to advance justification and finalization.
-                await self.sync_service.on_gossip_aggregated_attestation(att)
+                await self.sync_service.on_gossip_aggregated_attestation(att, peer_id)
 
             case PeerStatusEvent(peer_id=peer_id, status=status):
                 # Route peer status updates to sync service.

src/lean_spec/subspecs/node/node.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import signal
 import time
 from collections.abc import Callable
@@ -42,6 +43,11 @@
 from lean_spec.subspecs.validator import ValidatorRegistry, ValidatorService
 from lean_spec.types import Bytes32, Uint64
 
+logger = logging.getLogger(__name__)
+
+# Interval in seconds for periodic justified/finalized slot logging.
+_JUSTIFIED_FINALIZED_LOG_INTERVAL_SEC: Final = 10.0
+
 _ZERO_TIME: Final = Uint64(0)
 """Default genesis time for database loading when no genesis time is available."""
 
@@ -424,6 +430,7 @@ async def run(self, *, install_signal_handlers: bool = True) -> None:
                     tg.create_task(self.api_server.run())
                 if self.validator_service is not None:
                     tg.create_task(self.validator_service.run())
+                tg.create_task(self._log_justified_finalized_periodically())
                 tg.create_task(self._wait_shutdown())
         finally:
             if self.database is not None:
@@ -446,6 +453,36 @@ def _install_signal_handlers(self) -> None:
             # Cannot add handlers outside main thread.
             pass
 
+    async def _log_justified_finalized_periodically(self) -> None:
+        """
+        Log latest justified and finalized slot periodically.
+
+        Runs every _JUSTIFIED_FINALIZED_LOG_INTERVAL_SEC seconds to aid
+        monitoring and debugging of consensus progress.
+        """
+        while not self._shutdown.is_set():
+            await asyncio.sleep(_JUSTIFIED_FINALIZED_LOG_INTERVAL_SEC)
+            if self._shutdown.is_set():
+                break
+            store = self.sync_service.store
+            peers_connected = sum(
+                1 for p in self.sync_service.peer_manager.get_all_peers() if p.is_connected()
+            )
+            j = store.latest_justified
+            f = store.latest_finalized
+            j_root = j.root.hex() if hasattr(j.root, "hex") else str(j.root)
+            f_root = f.root.hex() if hasattr(f.root, "hex") else str(f.root)
+            logger.info("=" * 64)
+            logger.info(
+                "Peers=%s | Justified slot=%s root=%s | Finalized slot=%s root=%s",
+                peers_connected,
+                j.slot,
+                j_root,
+                f.slot,
+                f_root,
+            )
+            logger.info("=" * 64)
+
     async def _wait_shutdown(self) -> None:
         """
         Wait for shutdown signal then stop services.

src/lean_spec/subspecs/sync/service.py

@@ -420,7 +420,12 @@ async def on_gossip_block(
             )
             return
 
-        logger.debug("Processing gossip block from %s in state %s", peer_id, self._state.name)
+        logger.info(
+            "Block received from peer %s slot=%s (state=%s)",
+            peer_id,
+            block.message.block.slot,
+            self._state.name,
+        )
 
         if self._head_sync is None:
             raise RuntimeError("HeadSync not initialized")
@@ -440,6 +445,14 @@ async def on_gossip_block(
         #
         # A block may be cached instead of processed if its parent is unknown.
         if result.processed:
+            slot = block.message.block.slot
+            block_root = hash_tree_root(block.message.block)
+            logger.info(
+                "Block processed slot=%s root=%s from peer %s",
+                slot,
+                block_root.hex(),
+                peer_id,
+            )
             self.store = new_store
             self._replay_pending_attestations()
 
@@ -452,6 +465,7 @@ async def on_gossip_block(
     async def on_gossip_attestation(
         self,
         attestation: SignedAttestation,
+        peer_id: PeerId | None = None,
     ) -> None:
         """
         Handle attestation received via gossip.
@@ -465,6 +479,7 @@ async def on_gossip_attestation(
 
         Args:
             attestation: The signed attestation received.
+            peer_id: Peer that propagated the attestation (None if produced locally).
         """
         # Guard: Only process gossip in states that accept it.
         #
@@ -473,6 +488,16 @@ async def on_gossip_attestation(
         if not self._state.accepts_gossip:
             return
 
+        slot = attestation.data.slot
+        validator_id = attestation.validator_id
+        peer_str = str(peer_id) if peer_id is not None else "local"
+        logger.info(
+            "Attestation received from peer %s slot=%s validator=%s",
+            peer_str,
+            slot,
+            validator_id,
+        )
+
         # Check if we are an aggregator.
         #
         # A validator acts as an aggregator when it is active (has an ID)
@@ -489,7 +514,20 @@ async def on_gossip_attestation(
                 signed_attestation=attestation,
                 is_aggregator=is_aggregator_role,
             )
-        except (AssertionError, KeyError):
+            logger.info(
+                "Attestation from peer %s slot=%s validator=%s: validation and signature ok",
+                peer_str,
+                slot,
+                validator_id,
+            )
+        except (AssertionError, KeyError) as e:
+            logger.warning(
+                "Attestation from peer %s slot=%s validator=%s: validation or signature failed: %s",
+                peer_str,
+                slot,
+                validator_id,
+                e,
+            )
             # Attestation references a block not yet in our store.
             #
             # Buffer it for replay after the next block is processed.
@@ -502,6 +540,7 @@ async def on_gossip_attestation(
     async def on_gossip_aggregated_attestation(
         self,
         signed_attestation: SignedAggregatedAttestation,
+        peer_id: PeerId | None = None,
     ) -> None:
         """
         Handle aggregated attestation received via gossip.
@@ -512,13 +551,33 @@ async def on_gossip_aggregated_attestation(
 
         Args:
             signed_attestation: The signed aggregated attestation received.
+            peer_id: Peer that propagated the attestation (None if produced locally).
         """
         if not self._state.accepts_gossip:
             return
 
+        slot = signed_attestation.data.slot
+        peer_str = str(peer_id) if peer_id is not None else "local"
+        logger.info(
+            "Aggregated attestation received from peer %s slot=%s",
+            peer_str,
+            slot,
+        )
+
         try:
             self.store = self.store.on_gossip_aggregated_attestation(signed_attestation)
-        except (AssertionError, KeyError):
+            logger.info(
+                "Aggregated attestation from peer %s slot=%s: validation and signature ok",
+                peer_str,
+                slot,
+            )
+        except (AssertionError, KeyError) as e:
+            logger.warning(
+                "Aggregated attestation from peer %s slot=%s: validation or signature failed: %s",
+                peer_str,
+                slot,
+                e,
+            )
             # Target block not yet processed. Buffer for replay.
             self._pending_aggregated_attestations.append(signed_attestation)
             if len(self._pending_aggregated_attestations) > MAX_PENDING_ATTESTATIONS: