Skip to content

Commit 0dc7b27

Browse files
julienrbrttac0turtle
julienrbrt
and
tac0turtle
authored
fix(block/syncing): save data to p2p stores (#2736)
<!-- Please read and fill out this form before submitting your PR. Please make sure you have reviewed our contributors guide before submitting your first PR. NOTE: PR titles should follow semantic commits: https://www.conventionalcommits.org/en/v1.0.0/ --> ## Overview The syncing component was not saving header and event to the goheader (p2p) store. Which was an issue when peers were asking for heights. <!-- Please provide an explanation of the PR, including the appropriate context, background, goal, and rationale. If there is an issue with this information, please provide a tl;dr and link the issue. Ex: Closes #<issue number> --> --------- Co-authored-by: tac0turtle <marko@baricevic.me>
1 parent f09e382 commit 0dc7b27

19 files changed

Lines changed: 368 additions & 137 deletions

‎.mockery.yaml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,3 +58,15 @@ packages:
5858
dir: ./block/internal/syncing
5959
pkgname: syncing
6060
filename: syncer_mock.go
61+
github.com/evstack/ev-node/block/internal/common:
62+
interfaces:
63+
Broadcaster:
64+
config:
65+
dir: ./block/internal/common
66+
pkgname: common
67+
filename: broadcaster_mock.go
68+
p2pHandler:
69+
config:
70+
dir: ./block/internal/syncing
71+
pkgname: syncing
72+
filename: syncer_mock.go

‎block/components.go‎

Lines changed: 5 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,10 @@ import (
55
"errors"
66
"fmt"
77

8-
goheader "github.com/celestiaorg/go-header"
98
"github.com/rs/zerolog"
109

1110
"github.com/evstack/ev-node/block/internal/cache"
11+
"github.com/evstack/ev-node/block/internal/common"
1212
"github.com/evstack/ev-node/block/internal/executing"
1313
"github.com/evstack/ev-node/block/internal/reaping"
1414
"github.com/evstack/ev-node/block/internal/submitting"
@@ -122,11 +122,6 @@ func (bc *Components) Stop() error {
122122
return errs
123123
}
124124

125-
// broadcaster interface for P2P broadcasting
126-
type broadcaster[T any] interface {
127-
WriteToStoreAndBroadcast(ctx context.Context, payload T) error
128-
}
129-
130125
// NewSyncComponents creates components for a non-aggregator full node that can only sync blocks.
131126
// Non-aggregator full nodes can sync from P2P and DA but cannot produce blocks or submit to DA.
132127
// They have more sync capabilities than light nodes but no block production. No signer required.
@@ -136,8 +131,8 @@ func NewSyncComponents(
136131
store store.Store,
137132
exec coreexecutor.Executor,
138133
da coreda.DA,
139-
headerStore goheader.Store[*types.SignedHeader],
140-
dataStore goheader.Store[*types.Data],
134+
headerStore common.Broadcaster[*types.SignedHeader],
135+
dataStore common.Broadcaster[*types.Data],
141136
logger zerolog.Logger,
142137
metrics *Metrics,
143138
blockOpts BlockOptions,
@@ -199,8 +194,8 @@ func NewAggregatorComponents(
199194
sequencer coresequencer.Sequencer,
200195
da coreda.DA,
201196
signer signer.Signer,
202-
headerBroadcaster broadcaster[*types.SignedHeader],
203-
dataBroadcaster broadcaster[*types.Data],
197+
headerBroadcaster common.Broadcaster[*types.SignedHeader],
198+
dataBroadcaster common.Broadcaster[*types.Data],
204199
logger zerolog.Logger,
205200
metrics *Metrics,
206201
blockOpts BlockOptions,

‎block/internal/common/broadcaster_mock.go‎

Lines changed: 142 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎block/internal/common/event.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,22 @@ package common
22

33
import "github.com/evstack/ev-node/types"
44

5+
// EventSource represents the origin of a block event
6+
type EventSource string
7+
8+
const (
9+
// SourceDA indicates the event came from the DA layer
10+
SourceDA EventSource = "DA"
11+
// SourceP2P indicates the event came from P2P network
12+
SourceP2P EventSource = "P2P"
13+
)
14+
515
// DAHeightEvent represents a DA event for caching
616
type DAHeightEvent struct {
717
Header *types.SignedHeader
818
Data *types.Data
919
// DaHeight corresponds to the highest DA included height between the Header and Data.
1020
DaHeight uint64
21+
// Source indicates where this event originated from (DA or P2P)
22+
Source EventSource
1123
}

‎block/internal/common/expected_interfaces.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
package common
2+
3+
import (
4+
"context"
5+
6+
"github.com/celestiaorg/go-header"
7+
)
8+
9+
// broadcaster interface for P2P broadcasting
10+
type Broadcaster[H header.Header[H]] interface {
11+
WriteToStoreAndBroadcast(ctx context.Context, payload H) error
12+
Store() header.Store[H]
13+
}

‎block/internal/executing/executor.go‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,6 @@ import (
2424
"github.com/evstack/ev-node/types"
2525
)
2626

27-
// broadcaster interface for P2P broadcasting
28-
type broadcaster[T any] interface {
29-
WriteToStoreAndBroadcast(ctx context.Context, payload T) error
30-
}
31-
3227
// Executor handles block production, transaction processing, and state management
3328
type Executor struct {
3429
// Core components
@@ -42,8 +37,8 @@ type Executor struct {
4237
metrics *common.Metrics
4338

4439
// Broadcasting
45-
headerBroadcaster broadcaster[*types.SignedHeader]
46-
dataBroadcaster broadcaster[*types.Data]
40+
headerBroadcaster common.Broadcaster[*types.SignedHeader]
41+
dataBroadcaster common.Broadcaster[*types.Data]
4742

4843
// Configuration
4944
config config.Config
@@ -81,8 +76,8 @@ func NewExecutor(
8176
metrics *common.Metrics,
8277
config config.Config,
8378
genesis genesis.Genesis,
84-
headerBroadcaster broadcaster[*types.SignedHeader],
85-
dataBroadcaster broadcaster[*types.Data],
79+
headerBroadcaster common.Broadcaster[*types.SignedHeader],
80+
dataBroadcaster common.Broadcaster[*types.Data],
8681
logger zerolog.Logger,
8782
options common.BlockOptions,
8883
errorCh chan<- error,

‎block/internal/executing/executor_lazy_test.go‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -47,8 +47,10 @@ func TestLazyMode_ProduceBlockLogic(t *testing.T) {
4747

4848
mockExec := testmocks.NewMockExecutor(t)
4949
mockSeq := testmocks.NewMockSequencer(t)
50-
hb := &mockBroadcaster[*types.SignedHeader]{}
51-
db := &mockBroadcaster[*types.Data]{}
50+
hb := common.NewMockBroadcaster[*types.SignedHeader](t)
51+
hb.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
52+
db := common.NewMockBroadcaster[*types.Data](t)
53+
db.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
5254

5355
exec, err := NewExecutor(
5456
memStore,
@@ -155,8 +157,10 @@ func TestRegularMode_ProduceBlockLogic(t *testing.T) {
155157

156158
mockExec := testmocks.NewMockExecutor(t)
157159
mockSeq := testmocks.NewMockSequencer(t)
158-
hb := &mockBroadcaster[*types.SignedHeader]{}
159-
db := &mockBroadcaster[*types.Data]{}
160+
hb := common.NewMockBroadcaster[*types.SignedHeader](t)
161+
hb.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
162+
db := common.NewMockBroadcaster[*types.Data](t)
163+
db.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
160164

161165
exec, err := NewExecutor(
162166
memStore,

‎block/internal/executing/executor_logic_test.go‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -67,9 +67,11 @@ func TestProduceBlock_EmptyBatch_SetsEmptyDataHash(t *testing.T) {
6767
mockExec := testmocks.NewMockExecutor(t)
6868
mockSeq := testmocks.NewMockSequencer(t)
6969

70-
// Broadcasters are required by produceBlock; use simple mocks
71-
hb := &mockBroadcaster[*types.SignedHeader]{}
72-
db := &mockBroadcaster[*types.Data]{}
70+
// Broadcasters are required by produceBlock; use generated mocks
71+
hb := common.NewMockBroadcaster[*types.SignedHeader](t)
72+
hb.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
73+
db := common.NewMockBroadcaster[*types.Data](t)
74+
db.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
7375

7476
exec, err := NewExecutor(
7577
memStore,
@@ -126,8 +128,7 @@ func TestProduceBlock_EmptyBatch_SetsEmptyDataHash(t *testing.T) {
126128
assert.EqualValues(t, common.DataHashForEmptyTxs, sh.DataHash)
127129

128130
// Broadcasters should have been called with the produced header and data
129-
assert.True(t, hb.called)
130-
assert.True(t, db.called)
131+
// The testify mock framework tracks calls automatically
131132
}
132133

133134
func TestPendingLimit_SkipsProduction(t *testing.T) {
@@ -154,8 +155,10 @@ func TestPendingLimit_SkipsProduction(t *testing.T) {
154155

155156
mockExec := testmocks.NewMockExecutor(t)
156157
mockSeq := testmocks.NewMockSequencer(t)
157-
hb := &mockBroadcaster[*types.SignedHeader]{}
158-
db := &mockBroadcaster[*types.Data]{}
158+
hb := common.NewMockBroadcaster[*types.SignedHeader](t)
159+
hb.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
160+
db := common.NewMockBroadcaster[*types.Data](t)
161+
db.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
159162

160163
exec, err := NewExecutor(
161164
memStore,

0 commit comments

Comments
 (0)