fix(block/internal/syncing): only publish header/data locally if source.DA -- do not pollute p2p (#2777)

This is an attempt to fix subscriber too slow problem by only publishing
height events that come from DA layer to local subscription (which
prevents notifications to external peers). This will allow p2p syncer(s)
to see new heights + trigger sync loop to fill in gaps inside go-header
store(s) without polluting your p2p network.

Please review carefully and also please answer the `TODO` I left in
there as I still don't really understand the logic there.

---------

Co-authored-by: tac0turtle <[email protected]>

Author: renaynay

block/internal/common/broadcaster_mock.go

@@ -2,12 +2,13 @@ package common
 
 import (
 	"context"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 
 	"github.com/celestiaorg/go-header"
 )
 
 // broadcaster interface for P2P broadcasting
 type Broadcaster[H header.Header[H]] interface {
-	WriteToStoreAndBroadcast(ctx context.Context, payload H) error
+	WriteToStoreAndBroadcast(ctx context.Context, payload H, opts ...pubsub.PubOpt) error
 	Store() header.Store[H]
 }

block/internal/common/expected_interfaces.go

@@ -5,17 +5,19 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/evstack/ev-node/block/internal/cache"
-	"github.com/evstack/ev-node/block/internal/common"
 	coreda "github.com/evstack/ev-node/core/da"
 	coreexecutor "github.com/evstack/ev-node/core/execution"
+
+	"github.com/evstack/ev-node/block/internal/cache"
+	"github.com/evstack/ev-node/block/internal/common"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/store"
@@ -349,6 +351,8 @@ func (s *Syncer) tryFetchFromP2P() {
 
 	// Process data (if not already processed by headers)
 	newDataHeight := s.dataStore.Store().Height()
+	// TODO @MARKO: why only if newDataHeight != newHeaderHeight? why not process
+	//  just if newDataHeight > currentHeight ?
 	if newDataHeight != newHeaderHeight && newDataHeight > currentHeight {
 		s.p2pHandler.ProcessDataRange(s.ctx, currentHeight+1, newDataHeight, s.heightInCh)
 	}
@@ -415,8 +419,16 @@ func (s *Syncer) processHeightEvent(event *common.DAHeightEvent) {
 	// only save to p2p stores if the event came from DA
 	if event.Source == common.SourceDA {
 		g, ctx := errgroup.WithContext(s.ctx)
-		g.Go(func() error { return s.headerStore.WriteToStoreAndBroadcast(ctx, event.Header) })
-		g.Go(func() error { return s.dataStore.WriteToStoreAndBroadcast(ctx, event.Data) })
+		g.Go(func() error {
+			// broadcast header locally only — prevents spamming the p2p network with old height notifications,
+			// allowing the syncer to update its target and fill missing blocks
+			return s.headerStore.WriteToStoreAndBroadcast(ctx, event.Header, pubsub.WithLocalPublication(true))
+		})
+		g.Go(func() error {
+			// broadcast data locally only — prevents spamming the p2p network with old height notifications,
+			// allowing the syncer to update its target and fill missing blocks
+			return s.dataStore.WriteToStoreAndBroadcast(ctx, event.Data, pubsub.WithLocalPublication(true))
+		})
 		if err := g.Wait(); err != nil {
 			s.logger.Error().Err(err).Msg("failed to append event header and/or data to p2p store")
 		}

block/internal/syncing/syncer.go

@@ -123,7 +123,7 @@ func (syncService *SyncService[H]) Store() header.Store[H] {
 
 // WriteToStoreAndBroadcast initializes store if needed and broadcasts provided header or block.
 // Note: Only returns an error in case store can't be initialized. Logs error if there's one while broadcasting.
-func (syncService *SyncService[H]) WriteToStoreAndBroadcast(ctx context.Context, headerOrData H) error {
+func (syncService *SyncService[H]) WriteToStoreAndBroadcast(ctx context.Context, headerOrData H, opts ...pubsub.PubOpt) error {
 	if syncService.genesis.InitialHeight == 0 {
 		return fmt.Errorf("invalid initial height; cannot be zero")
 	}
@@ -148,7 +148,7 @@ func (syncService *SyncService[H]) WriteToStoreAndBroadcast(ctx context.Context,
 	}
 
 	// Broadcast for subscribers
-	if err := syncService.sub.Broadcast(ctx, headerOrData); err != nil {
+	if err := syncService.sub.Broadcast(ctx, headerOrData, opts...); err != nil {
 		// for the first block when starting the app, broadcast error is expected
 		// as we have already initialized the store for starting the syncer.
 		// Hence, we ignore the error. Exact reason: validation ignored

pkg/sync/sync_service.go

@@ -162,6 +162,9 @@ func (h *Header) ToProto() *pb.Header {
 		ChainId:         h.BaseHeader.ChainID,
 		ValidatorHash:   h.ValidatorHash,
 	}
+	if unknown := encodeLegacyUnknownFields(h.Legacy); len(unknown) > 0 {
+		pHeader.ProtoReflect().SetUnknown(unknown)
+	}
 	return pHeader
 }
 
@@ -509,6 +512,28 @@ func decodeLegacyHeaderFields(pHeader *pb.Header) (*LegacyHeaderFields, error) {
 	return &legacy, nil
 }
 
+func encodeLegacyUnknownFields(legacy *LegacyHeaderFields) []byte {
+	if legacy == nil || legacy.IsZero() {
+		return nil
+	}
+
+	var payload []byte
+
+	if len(legacy.LastCommitHash) > 0 {
+		payload = appendBytesField(payload, legacyLastCommitHashField, legacy.LastCommitHash)
+	}
+
+	if len(legacy.ConsensusHash) > 0 {
+		payload = appendBytesField(payload, legacyConsensusHashField, legacy.ConsensusHash)
+	}
+
+	if len(legacy.LastResultsHash) > 0 {
+		payload = appendBytesField(payload, legacyLastResultsHashField, legacy.LastResultsHash)
+	}
+
+	return payload
+}
+
 func appendBytesField(buf []byte, number protowire.Number, value []byte) []byte {
 	buf = protowire.AppendTag(buf, number, protowire.BytesType)
 	buf = protowire.AppendVarint(buf, uint64(len(value)))

types/serialization.go

@@ -1,6 +1,7 @@
 package types
 
 import (
+	"bytes"
 	"crypto/rand"
 	"testing"
 	"time"
@@ -355,6 +356,53 @@ func TestHeader_HashFields_NilAndEmpty(t *testing.T) {
 	assert.Nil(t, h2.ValidatorHash)
 }
 
+func TestHeaderMarshalBinary_PreservesLegacyFields(t *testing.T) {
+	t.Parallel()
+
+	header := &Header{
+		Version:    Version{Block: 2, App: 3},
+		BaseHeader: BaseHeader{Height: 42, Time: 123456789, ChainID: "chain-legacy"},
+		LastHeaderHash: []byte{
+			0x01, 0x02, 0x03, 0x04,
+		},
+		DataHash:        []byte{0x05, 0x06, 0x07, 0x08},
+		AppHash:         []byte{0x09, 0x0A, 0x0B, 0x0C},
+		ProposerAddress: []byte{0x0D, 0x0E, 0x0F, 0x10},
+		ValidatorHash:   []byte{0x11, 0x12, 0x13, 0x14},
+		Legacy: &LegacyHeaderFields{
+			LastCommitHash:  bytes.Repeat([]byte{0x21}, 32),
+			ConsensusHash:   bytes.Repeat([]byte{0x22}, 32),
+			LastResultsHash: bytes.Repeat([]byte{0x23}, 32),
+		},
+	}
+
+	unknown := header.ToProto().ProtoReflect().GetUnknown()
+	require.NotEmpty(t, unknown, "legacy fields should be serialized into unknown proto fields")
+
+	raw, err := header.MarshalBinary()
+	require.NoError(t, err)
+	require.NotEmpty(t, raw)
+
+	var decoded Header
+	require.NoError(t, decoded.UnmarshalBinary(raw))
+	require.NotNil(t, decoded.Legacy)
+	require.False(t, decoded.Legacy.IsZero())
+
+	assert.Equal(t, header.Legacy.LastCommitHash, decoded.Legacy.LastCommitHash)
+	assert.Equal(t, header.Legacy.ConsensusHash, decoded.Legacy.ConsensusHash)
+	assert.Equal(t, header.Legacy.LastResultsHash, decoded.Legacy.LastResultsHash)
+
+	origHash := header.Hash()
+	decodedHash := decoded.Hash()
+	require.NotNil(t, origHash)
+	require.NotNil(t, decodedHash)
+	assert.Equal(t, origHash, decodedHash)
+
+	legacyHash, err := header.HashLegacy()
+	require.NoError(t, err)
+	assert.Equal(t, legacyHash, origHash)
+}
+
 // TestProtoConversionConsistency_AllTypes checks that ToProto/FromProto are true inverses for all major types.
 func TestProtoConversionConsistency_AllTypes(t *testing.T) {
 	// Header