add p2p logs to help debug issue

Author: tac0turtle

block/internal/syncing/syncer.go

@@ -42,6 +42,8 @@ const (
 	fullnessThreshold = 0.8
 )
 
+var p2pWaitLogInterval = 5 * time.Second
+
 // Syncer handles block synchronization from DA and P2P sources.
 type Syncer struct {
 	// Core components
@@ -476,7 +478,7 @@ func (s *Syncer) p2pWorkerLoop(ctx context.Context) {
 		waitCtx, cancel := context.WithCancel(ctx)
 		s.setP2PWaitState(targetHeight, cancel)
 
-		err = s.p2pHandler.ProcessHeight(waitCtx, targetHeight, s.heightInCh)
+		err = s.processP2PHeight(waitCtx, targetHeight, logger)
 		s.cancelP2PWait(targetHeight)
 
 		if err != nil {
@@ -503,6 +505,54 @@ func (s *Syncer) p2pWorkerLoop(ctx context.Context) {
 	}
 }
 
+func (s *Syncer) processP2PHeight(ctx context.Context, targetHeight uint64, logger zerolog.Logger) error {
+	doneCh := make(chan error, 1)
+	startedAt := time.Now()
+	loggedWait := false
+
+	go func() {
+		doneCh <- s.p2pHandler.ProcessHeight(ctx, targetHeight, s.heightInCh)
+	}()
+
+	ticker := time.NewTicker(p2pWaitLogInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case err := <-doneCh:
+			if loggedWait && err == nil {
+				logger.Info().
+					Uint64("height", targetHeight).
+					Dur("waited", time.Since(startedAt)).
+					Msg("P2P height became available")
+			}
+			return err
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			loggedWait = true
+
+			event := logger.Warn().
+				Uint64("target_height", targetHeight).
+				Uint64("header_store_height", storeHeightOf(s.headerStore)).
+				Uint64("data_store_height", storeHeightOf(s.dataStore)).
+				Int("height_events_buffered", len(s.heightInCh)).
+				Int64("in_flight_events", s.inFlight.Load()).
+				Int("pending_events", s.cache.PendingEventsCount()).
+				Int("total_pending", s.PendingCount()).
+				Dur("waited", time.Since(startedAt))
+
+			if localHeight, err := s.store.Height(ctx); err != nil {
+				event = event.Str("local_height_err", err.Error())
+			} else {
+				event = event.Uint64("local_height", localHeight)
+			}
+
+			event.Msg("waiting for P2P height to become available")
+		}
+	}
+}
+
 func (s *Syncer) waitForGenesis() bool {
 	if delay := time.Until(s.genesis.StartTime); delay > 0 {
 		timer := time.NewTimer(delay)
@@ -1110,6 +1160,17 @@ func (s *Syncer) sleepOrDone(ctx context.Context, duration time.Duration) bool {
 	}
 }
 
+type storeHeightReader interface {
+	Height() uint64
+}
+
+func storeHeightOf(store storeHeightReader) uint64 {
+	if store == nil {
+		return 0
+	}
+	return store.Height()
+}
+
 type p2pWaitState struct {
 	height uint64
 	cancel context.CancelFunc

block/internal/syncing/syncer_test.go

@@ -1,11 +1,13 @@
 package syncing
 
 import (
+	"bytes"
 	"context"
 	crand "crypto/rand"
 	"crypto/sha512"
 	"errors"
 	"math"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"testing/synctest"
@@ -34,6 +36,16 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
+type stubP2PHandler struct {
+	processHeight func(ctx context.Context, height uint64, heightInCh chan<- common.DAHeightEvent) error
+}
+
+func (s *stubP2PHandler) ProcessHeight(ctx context.Context, height uint64, heightInCh chan<- common.DAHeightEvent) error {
+	return s.processHeight(ctx, height, heightInCh)
+}
+
+func (s *stubP2PHandler) SetProcessedHeight(uint64) {}
+
 // helper to create a signer, pubkey and address for tests
 func buildSyncTestSigner(tb testing.TB) (addr []byte, pub crypto.PubKey, signer signerpkg.Signer) {
 	tb.Helper()
@@ -165,6 +177,65 @@ func TestSyncer_validateBlock_DataHashMismatch(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestSyncer_processP2PHeight_LogsWhenBlocked(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+
+	batch, err := st.NewBatch(t.Context())
+	require.NoError(t, err)
+	require.NoError(t, batch.SetHeight(7))
+	require.NoError(t, batch.Commit())
+
+	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	require.NoError(t, err)
+	cm.SetPendingEvent(10, &common.DAHeightEvent{})
+
+	headerStore := extmocks.NewMockStore[*types.P2PSignedHeader](t)
+	headerStore.EXPECT().Height().Return(uint64(11)).Maybe()
+	dataStore := extmocks.NewMockStore[*types.P2PData](t)
+	dataStore.EXPECT().Height().Return(uint64(12)).Maybe()
+
+	var logBuf bytes.Buffer
+	logger := zerolog.New(&logBuf)
+
+	originalInterval := p2pWaitLogInterval
+	p2pWaitLogInterval = 10 * time.Millisecond
+	t.Cleanup(func() {
+		p2pWaitLogInterval = originalInterval
+	})
+
+	s := &Syncer{
+		store:       st,
+		cache:       cm,
+		headerStore: headerStore,
+		dataStore:   dataStore,
+		heightInCh:  make(chan common.DAHeightEvent, 2),
+		logger:      logger,
+		p2pHandler: &stubP2PHandler{
+			processHeight: func(ctx context.Context, _ uint64, _ chan<- common.DAHeightEvent) error {
+				select {
+				case <-time.After(35 * time.Millisecond):
+					return nil
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			},
+		},
+	}
+
+	err = s.processP2PHeight(t.Context(), 8, logger)
+	require.NoError(t, err)
+
+	logs := logBuf.String()
+	require.Contains(t, logs, "waiting for P2P height to become available")
+	require.Contains(t, logs, "\"target_height\":8")
+	require.Contains(t, logs, "\"local_height\":7")
+	require.Contains(t, logs, "\"header_store_height\":11")
+	require.Contains(t, logs, "\"data_store_height\":12")
+	require.Contains(t, logs, "\"pending_events\":1")
+	require.True(t, strings.Contains(logs, "P2P height became available"))
+}
+
 func TestProcessHeightEvent_SyncsAndUpdatesState(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)