1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
package p2p
import (
"context"
"math/big"
"testing"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/peer"
mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
"github.com/stretchr/testify/require"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum-optimism/optimism/op-node/eth"
"github.com/ethereum-optimism/optimism/op-node/metrics"
"github.com/ethereum-optimism/optimism/op-node/rollup"
"github.com/ethereum-optimism/optimism/op-node/testlog"
)
type mockPayloadFn func(n uint64) (*eth.ExecutionPayload, error)
func (fn mockPayloadFn) PayloadByNumber(_ context.Context, number uint64) (*eth.ExecutionPayload, error) {
return fn(number)
}
var _ L2Chain = mockPayloadFn(nil)
func setupSyncTestData(length uint64) (*rollup.Config, map[uint64]*eth.ExecutionPayload, func(i uint64) eth.L2BlockRef) {
// minimal rollup config to build mock blocks & verify their time.
cfg := &rollup.Config{
Genesis: rollup.Genesis{
L1: eth.BlockID{Hash: common.Hash{0xaa}},
L2: eth.BlockID{Hash: common.Hash{0xbb}},
L2Time: 9000,
},
BlockTime: 2,
L2ChainID: big.NewInt(1234),
}
// create some simple fake test blocks
payloads := make(map[uint64]*eth.ExecutionPayload)
payloads[0] = ð.ExecutionPayload{
Timestamp: eth.Uint64Quantity(cfg.Genesis.L2Time),
}
payloads[0].BlockHash, _ = payloads[0].CheckBlockHash()
for i := uint64(1); i <= length; i++ {
payload := ð.ExecutionPayload{
ParentHash: payloads[i-1].BlockHash,
BlockNumber: eth.Uint64Quantity(i),
Timestamp: eth.Uint64Quantity(cfg.Genesis.L2Time + i*cfg.BlockTime),
}
payload.BlockHash, _ = payload.CheckBlockHash()
payloads[i] = payload
}
l2Ref := func(i uint64) eth.L2BlockRef {
return eth.L2BlockRef{
Hash: payloads[i].BlockHash,
Number: uint64(payloads[i].BlockNumber),
ParentHash: payloads[i].ParentHash,
Time: uint64(payloads[i].Timestamp),
}
}
return cfg, payloads, l2Ref
}
func TestSinglePeerSync(t *testing.T) {
t.Parallel() // Takes a while, but can run in parallel
log := testlog.Logger(t, log.LvlError)
cfg, payloads, l2Ref := setupSyncTestData(25)
// Serving payloads: just load them from the map, if they exist
servePayload := mockPayloadFn(func(n uint64) (*eth.ExecutionPayload, error) {
p, ok := payloads[n]
if !ok {
return nil, ethereum.NotFound
}
return p, nil
})
// collect received payloads in a buffered channel, so we can verify we get everything
received := make(chan *eth.ExecutionPayload, 100)
receivePayload := receivePayloadFn(func(ctx context.Context, from peer.ID, payload *eth.ExecutionPayload) error {
received <- payload
return nil
})
// Setup 2 minimal test hosts to attach the sync protocol to
mnet, err := mocknet.FullMeshConnected(2)
require.NoError(t, err, "failed to setup mocknet")
defer mnet.Close()
hosts := mnet.Hosts()
hostA, hostB := hosts[0], hosts[1]
require.Equal(t, hostA.Network().Connectedness(hostB.ID()), network.Connected)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Setup host A as the server
srv := NewReqRespServer(cfg, servePayload, metrics.NoopMetrics)
payloadByNumber := MakeStreamHandler(ctx, log.New("role", "server"), srv.HandleSyncRequest)
hostA.SetStreamHandler(PayloadByNumberProtocolID(cfg.L2ChainID), payloadByNumber)
// Setup host B as the client
cl := NewSyncClient(log.New("role", "client"), cfg, hostB.NewStream, receivePayload, metrics.NoopMetrics)
// Setup host B (client) to sync from its peer Host A (server)
cl.AddPeer(hostA.ID())
cl.Start()
defer cl.Close()
// request to start syncing between 10 and 20
require.NoError(t, cl.RequestL2Range(ctx, l2Ref(10), l2Ref(20)))
// and wait for the sync results to come in (in reverse order)
for i := uint64(19); i > 10; i-- {
p := <-received
require.Equal(t, uint64(p.BlockNumber), i, "expecting payloads in order")
exp, ok := payloads[uint64(p.BlockNumber)]
require.True(t, ok, "expecting known payload")
require.Equal(t, exp.BlockHash, p.BlockHash, "expecting the correct payload")
}
}
func TestMultiPeerSync(t *testing.T) {
t.Parallel() // Takes a while, but can run in parallel
log := testlog.Logger(t, log.LvlError)
cfg, payloads, l2Ref := setupSyncTestData(100)
setupPeer := func(ctx context.Context, h host.Host) (*SyncClient, chan *eth.ExecutionPayload) {
// Serving payloads: just load them from the map, if they exist
servePayload := mockPayloadFn(func(n uint64) (*eth.ExecutionPayload, error) {
p, ok := payloads[n]
if !ok {
return nil, ethereum.NotFound
}
return p, nil
})
// collect received payloads in a buffered channel, so we can verify we get everything
received := make(chan *eth.ExecutionPayload, 100)
receivePayload := receivePayloadFn(func(ctx context.Context, from peer.ID, payload *eth.ExecutionPayload) error {
received <- payload
return nil
})
// Setup as server
srv := NewReqRespServer(cfg, servePayload, metrics.NoopMetrics)
payloadByNumber := MakeStreamHandler(ctx, log.New("serve", "payloads_by_number"), srv.HandleSyncRequest)
h.SetStreamHandler(PayloadByNumberProtocolID(cfg.L2ChainID), payloadByNumber)
cl := NewSyncClient(log.New("role", "client"), cfg, h.NewStream, receivePayload, metrics.NoopMetrics)
return cl, received
}
// Setup 3 minimal test hosts to attach the sync protocol to
mnet, err := mocknet.FullMeshConnected(3)
require.NoError(t, err, "failed to setup mocknet")
defer mnet.Close()
hosts := mnet.Hosts()
hostA, hostB, hostC := hosts[0], hosts[1], hosts[2]
require.Equal(t, hostA.Network().Connectedness(hostB.ID()), network.Connected)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
clA, recvA := setupPeer(ctx, hostA)
clB, recvB := setupPeer(ctx, hostB)
clC, _ := setupPeer(ctx, hostC)
// Make them all sync from each other
clA.AddPeer(hostB.ID())
clA.AddPeer(hostC.ID())
clA.Start()
defer clA.Close()
clB.AddPeer(hostA.ID())
clB.AddPeer(hostC.ID())
clB.Start()
defer clB.Close()
clC.AddPeer(hostA.ID())
clC.AddPeer(hostB.ID())
clC.Start()
defer clC.Close()
// request to start syncing between 10 and 100
require.NoError(t, clA.RequestL2Range(ctx, l2Ref(10), l2Ref(90)))
// With such large range to request we are going to hit the rate-limits of B and C,
// but that means we'll balance the work between the peers.
p := <-recvA
exp, ok := payloads[uint64(p.BlockNumber)]
require.True(t, ok, "expecting known payload")
require.Equal(t, exp.BlockHash, p.BlockHash, "expecting the correct payload")
// now see if B can sync a range, and fill the gap with a re-request
bl25 := payloads[25] // temporarily remove it from the available payloads. This will create a gap
delete(payloads, uint64(25))
require.NoError(t, clB.RequestL2Range(ctx, l2Ref(20), l2Ref(30)))
for i := uint64(29); i > 25; i-- {
p := <-recvB
exp, ok := payloads[uint64(p.BlockNumber)]
require.True(t, ok, "expecting known payload")
require.Equal(t, exp.BlockHash, p.BlockHash, "expecting the correct payload")
}
// the request for 25 should fail. See:
// server: WARN peer requested unknown block by number num=25
// client: WARN failed p2p sync request num=25 err="peer failed to serve request with code 1"
require.Zero(t, len(recvB), "there is a gap, should not see other payloads yet")
// Add back the block
payloads[25] = bl25
// And request a range again, 25 is there now, and 21-24 should follow quickly (some may already have been fetched and wait in quarantine)
require.NoError(t, clB.RequestL2Range(ctx, l2Ref(20), l2Ref(26)))
for i := uint64(25); i > 20; i-- {
p := <-recvB
exp, ok := payloads[uint64(p.BlockNumber)]
require.True(t, ok, "expecting known payload")
require.Equal(t, exp.BlockHash, p.BlockHash, "expecting the correct payload")
}
}