hermez-node/coordinator/coordinator_test.go
coordinator/coordinator_test.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "io/ioutil"
7 "math/big"
8 "os"
9 "sync"
10 "testing"
11 "time"
12
13 ethCommon "github.com/ethereum/go-ethereum/common"
14 "github.com/hermeznetwork/hermez-node/batchbuilder"
15 "github.com/hermeznetwork/hermez-node/common"
16 dbUtils "github.com/hermeznetwork/hermez-node/db"
17 "github.com/hermeznetwork/hermez-node/db/historydb"
18 "github.com/hermeznetwork/hermez-node/db/l2db"
19 "github.com/hermeznetwork/hermez-node/db/statedb"
20 "github.com/hermeznetwork/hermez-node/eth"
21 "github.com/hermeznetwork/hermez-node/log"
22 "github.com/hermeznetwork/hermez-node/prover"
23 "github.com/hermeznetwork/hermez-node/synchronizer"
24 "github.com/hermeznetwork/hermez-node/test"
25 "github.com/hermeznetwork/hermez-node/test/til"
26 "github.com/hermeznetwork/hermez-node/txselector"
27 "github.com/hermeznetwork/tracerr"
28 "github.com/iden3/go-merkletree/db/pebble"
29 "github.com/stretchr/testify/assert"
30 "github.com/stretchr/testify/require"
31)
32
33var deleteme = []string{}
34
35func pebbleMakeCheckpoint(source, dest string) error {
36 // Remove dest folder (if it exists) before doing the checkpoint
37 if _, err := os.Stat(dest); !os.IsNotExist(err) {
38 err := os.RemoveAll(dest)
39 if err != nil {
40 return tracerr.Wrap(err)
41 }
42 } else if err != nil && !os.IsNotExist(err) {
43 return tracerr.Wrap(err)
44 }
45
46 sto, err := pebble.NewPebbleStorage(source, false)
47 if err != nil {
48 return tracerr.Wrap(err)
49 }
50 defer func() {
51 errClose := sto.Pebble().Close()
52 if errClose != nil {
53 log.Errorw("Pebble.Close", "err", errClose)
54 }
55 }()
56
57 // execute Checkpoint
58 err = sto.Pebble().Checkpoint(dest)
59 if err != nil {
60 return tracerr.Wrap(err)
61 }
62
63 return nil
64}
65
66func TestMain(m *testing.M) {
67 exitVal := m.Run()
68 for _, dir := range deleteme {
69 if err := os.RemoveAll(dir); err != nil {
70 panic(err)
71 }
72 }
73 os.Exit(exitVal)
74}
75
76var syncDBPath string
77var txSelDBPath string
78var batchBuilderDBPath string
79
80type modules struct {
81 historyDB *historydb.HistoryDB
82 l2DB *l2db.L2DB
83 txSelector *txselector.TxSelector
84 batchBuilder *batchbuilder.BatchBuilder
85 stateDB *statedb.StateDB
86}
87
88var maxL1UserTxs uint64 = 128
89var maxL1Txs uint64 = 256
90var maxL1CoordinatorTxs uint64 = maxL1Txs - maxL1UserTxs
91var maxTxs uint64 = 376
92var nLevels uint32 = 32 //nolint:deadcode,unused
93var maxFeeTxs uint32 = 64 //nolint:deadcode,varcheck
94var chainID uint16 = 0
95
96func newTestModules(t *testing.T) modules {
97 var err error
98 syncDBPath, err = ioutil.TempDir("", "tmpSyncDB")
99 require.NoError(t, err)
100 deleteme = append(deleteme, syncDBPath)
101 syncStateDB, err := statedb.NewStateDB(syncDBPath, statedb.TypeSynchronizer, 48, chainID)
102 assert.NoError(t, err)
103
104 pass := os.Getenv("POSTGRES_PASS")
105 db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez")
106 require.NoError(t, err)
107 test.WipeDB(db)
108 l2DB := l2db.NewL2DB(db, 10, 100, 24*time.Hour)
109 historyDB := historydb.NewHistoryDB(db)
110
111 txSelDBPath, err = ioutil.TempDir("", "tmpTxSelDB")
112 require.NoError(t, err)
113 deleteme = append(deleteme, txSelDBPath)
114
115 coordAccount := &txselector.CoordAccount{ // TODO TMP
116 Addr: ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
117 BJJ: common.EmptyBJJComp,
118 AccountCreationAuth: nil,
119 }
120 txSelector, err := txselector.NewTxSelector(coordAccount, txSelDBPath, syncStateDB, l2DB)
121 assert.NoError(t, err)
122
123 batchBuilderDBPath, err = ioutil.TempDir("", "tmpBatchBuilderDB")
124 require.NoError(t, err)
125 deleteme = append(deleteme, batchBuilderDBPath)
126 batchBuilder, err := batchbuilder.NewBatchBuilder(batchBuilderDBPath, syncStateDB, nil, 0, uint64(nLevels))
127 assert.NoError(t, err)
128
129 return modules{
130 historyDB: historyDB,
131 l2DB: l2DB,
132 txSelector: txSelector,
133 batchBuilder: batchBuilder,
134 stateDB: syncStateDB,
135 }
136}
137
138type timer struct {
139 time int64
140}
141
142func (t *timer) Time() int64 {
143 currentTime := t.time
144 t.time++
145 return currentTime
146}
147
148var bidder = ethCommon.HexToAddress("0x6b175474e89094c44da98b954eedeac495271d0f")
149var forger = ethCommon.HexToAddress("0xc344E203a046Da13b0B4467EB7B3629D0C99F6E6")
150
151func newTestCoordinator(t *testing.T, forgerAddr ethCommon.Address, ethClient *test.Client,
152 ethClientSetup *test.ClientSetup, modules modules) *Coordinator {
153 debugBatchPath, err := ioutil.TempDir("", "tmpDebugBatch")
154 require.NoError(t, err)
155 deleteme = append(deleteme, debugBatchPath)
156
157 conf := Config{
158 ForgerAddress: forgerAddr,
159 ConfirmBlocks: 5,
160 L1BatchTimeoutPerc: 0.5,
161 EthClientAttempts: 5,
162 SyncRetryInterval: 400 * time.Microsecond,
163 EthClientAttemptsDelay: 100 * time.Millisecond,
164 TxManagerCheckInterval: 300 * time.Millisecond,
165 DebugBatchPath: debugBatchPath,
166 Purger: PurgerCfg{
167 PurgeBatchDelay: 10,
168 PurgeBlockDelay: 10,
169 InvalidateBatchDelay: 4,
170 InvalidateBlockDelay: 4,
171 },
172 }
173 serverProofs := []prover.Client{
174 &prover.MockClient{Delay: 300 * time.Millisecond},
175 &prover.MockClient{Delay: 400 * time.Millisecond},
176 }
177
178 scConsts := &synchronizer.SCConsts{
179 Rollup: *ethClientSetup.RollupConstants,
180 Auction: *ethClientSetup.AuctionConstants,
181 WDelayer: *ethClientSetup.WDelayerConstants,
182 }
183 initSCVars := &synchronizer.SCVariables{
184 Rollup: *ethClientSetup.RollupVariables,
185 Auction: *ethClientSetup.AuctionVariables,
186 WDelayer: *ethClientSetup.WDelayerVariables,
187 }
188 coord, err := NewCoordinator(conf, modules.historyDB, modules.l2DB, modules.txSelector,
189 modules.batchBuilder, serverProofs, ethClient, scConsts, initSCVars)
190 require.NoError(t, err)
191 return coord
192}
193
194func newTestSynchronizer(t *testing.T, ethClient *test.Client, ethClientSetup *test.ClientSetup,
195 modules modules) *synchronizer.Synchronizer {
196 sync, err := synchronizer.NewSynchronizer(ethClient, modules.historyDB, modules.stateDB,
197 synchronizer.Config{
198 StatsRefreshPeriod: 0 * time.Second,
199 })
200 require.NoError(t, err)
201 return sync
202}
203
204// TestCoordinatorFlow is a test where the coordinator is stared (which means
205// that goroutines are spawned), and ethereum blocks are mined via the
206// test.Client to simulate starting and stopping forging times. This test
207// works without a synchronizer, and no l2txs are inserted in the pool, so all
208// the batches are forged empty. The purpose of this test is to manually
209// observe via the logs that nothing crashes and that the coordinator starts
210// and stops forging at the right blocks.
211func TestCoordinatorFlow(t *testing.T) {
212 if os.Getenv("TEST_COORD_FLOW") == "" {
213 return
214 }
215 ethClientSetup := test.NewClientSetupExample()
216 var timer timer
217 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
218 modules := newTestModules(t)
219 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
220
221 // Bid for slot 2 and 4
222 _, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
223 require.NoError(t, err)
224 _, err = ethClient.AuctionBidSimple(2, big.NewInt(9999))
225 require.NoError(t, err)
226 _, err = ethClient.AuctionBidSimple(4, big.NewInt(9999))
227 require.NoError(t, err)
228
229 coord.Start()
230 time.Sleep(1 * time.Second)
231
232 waitForSlot := func(slot int64) {
233 for {
234 blockNum, err := ethClient.EthLastBlock()
235 require.NoError(t, err)
236 nextBlockSlot, err := ethClient.AuctionGetSlotNumber(blockNum + 1)
237 require.NoError(t, err)
238 if nextBlockSlot == slot {
239 break
240 }
241 ethClient.CtlMineBlock()
242 time.Sleep(100 * time.Millisecond)
243 var stats synchronizer.Stats
244 stats.Eth.LastBlock = *ethClient.CtlLastBlock()
245 stats.Sync.LastBlock = stats.Eth.LastBlock
246 stats.Eth.LastBatch = ethClient.CtlLastForgedBatch()
247 stats.Sync.LastBatch = stats.Eth.LastBatch
248 canForge, err := ethClient.AuctionCanForge(forger, blockNum+1)
249 require.NoError(t, err)
250 if canForge {
251 // fmt.Println("DBG canForge")
252 stats.Sync.Auction.CurrentSlot.Forger = forger
253 }
254 // Copy stateDB to synchronizer if there was a new batch
255 source := fmt.Sprintf("%v/BatchNum%v", batchBuilderDBPath, stats.Sync.LastBatch)
256 dest := fmt.Sprintf("%v/BatchNum%v", syncDBPath, stats.Sync.LastBatch)
257 if stats.Sync.LastBatch != 0 {
258 if _, err := os.Stat(dest); os.IsNotExist(err) {
259 log.Infow("Making pebble checkpoint for sync",
260 "source", source, "dest", dest)
261 err = pebbleMakeCheckpoint(source, dest)
262 require.NoError(t, err)
263 }
264 }
265 coord.SendMsg(MsgSyncBlock{
266 Stats: stats,
267 })
268 }
269 }
270
271 // NOTE: With the current test, the coordinator will enter in forge
272 // time before the bidded slot because no one else is forging in the
273 // other slots before the slot deadline.
274 // simulate forgeSequence time
275 waitForSlot(2)
276 log.Info("~~~ simulate entering in forge time")
277 time.Sleep(1 * time.Second)
278
279 // simulate going out from forgeSequence
280 waitForSlot(3)
281 log.Info("~~~ simulate going out from forge time")
282 time.Sleep(1 * time.Second)
283
284 // simulate entering forgeSequence time again
285 waitForSlot(4)
286 log.Info("~~~ simulate entering in forge time again")
287 time.Sleep(2 * time.Second)
288
289 // simulate stopping forgerLoop by channel
290 log.Info("~~~ simulate stopping forgerLoop by closing coordinator stopch")
291 coord.Stop()
292 time.Sleep(1 * time.Second)
293}
294
295func TestCoordinatorStartStop(t *testing.T) {
296 ethClientSetup := test.NewClientSetupExample()
297 var timer timer
298 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
299 modules := newTestModules(t)
300 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
301 coord.Start()
302 coord.Stop()
303}
304
305func TestCoordCanForge(t *testing.T) {
306 ethClientSetup := test.NewClientSetupExample()
307 bootForger := ethClientSetup.AuctionVariables.BootCoordinator
308
309 var timer timer
310 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
311 modules := newTestModules(t)
312 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
313 _, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
314 require.NoError(t, err)
315 bid, ok := new(big.Int).SetString("12000000000000000000", 10)
316 if !ok {
317 panic("bad bid")
318 }
319 _, err = ethClient.AuctionBidSimple(2, bid)
320 require.NoError(t, err)
321
322 modules2 := newTestModules(t)
323 bootCoord := newTestCoordinator(t, bootForger, ethClient, ethClientSetup, modules2)
324
325 assert.Equal(t, forger, coord.cfg.ForgerAddress)
326 assert.Equal(t, bootForger, bootCoord.cfg.ForgerAddress)
327 ethBootCoord, err := ethClient.AuctionGetBootCoordinator()
328 require.NoError(t, err)
329 assert.Equal(t, &bootForger, ethBootCoord)
330
331 var stats synchronizer.Stats
332
333 // Slot 0. No bid, so the winner is the boot coordinator
334 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
335 stats.Sync.LastBlock = stats.Eth.LastBlock
336 stats.Sync.Auction.CurrentSlot.Forger = bootForger
337 assert.Equal(t, false, coord.canForge(&stats))
338 assert.Equal(t, true, bootCoord.canForge(&stats))
339
340 // Slot 0. No bid, and we reach the deadline, so anyone can forge
341 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
342 int64(ethClientSetup.AuctionVariables.SlotDeadline)
343 stats.Sync.LastBlock = stats.Eth.LastBlock
344 stats.Sync.Auction.CurrentSlot.Forger = bootForger
345 assert.Equal(t, true, coord.canForge(&stats))
346 assert.Equal(t, true, bootCoord.canForge(&stats))
347
348 // Slot 1. coordinator bid, so the winner is the coordinator
349 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
350 1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
351 stats.Sync.LastBlock = stats.Eth.LastBlock
352 stats.Sync.Auction.CurrentSlot.Forger = forger
353 assert.Equal(t, true, coord.canForge(&stats))
354 assert.Equal(t, false, bootCoord.canForge(&stats))
355}
356
357func TestCoordHandleMsgSyncBlock(t *testing.T) {
358 ethClientSetup := test.NewClientSetupExample()
359 bootForger := ethClientSetup.AuctionVariables.BootCoordinator
360
361 var timer timer
362 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
363 modules := newTestModules(t)
364 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
365 _, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
366 require.NoError(t, err)
367 bid, ok := new(big.Int).SetString("11000000000000000000", 10)
368 if !ok {
369 panic("bad bid")
370 }
371 _, err = ethClient.AuctionBidSimple(2, bid)
372 require.NoError(t, err)
373
374 var msg MsgSyncBlock
375 stats := &msg.Stats
376 ctx := context.Background()
377
378 // Slot 0. No bid, so the winner is the boot coordinator
379 // pipelineStarted: false -> false
380 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
381 stats.Sync.LastBlock = stats.Eth.LastBlock
382 stats.Sync.Auction.CurrentSlot.Forger = bootForger
383 assert.Equal(t, false, coord.canForge(stats))
384 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
385 assert.Nil(t, coord.pipeline)
386
387 // Slot 0. No bid, and we reach the deadline, so anyone can forge
388 // pipelineStarted: false -> true
389 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
390 int64(ethClientSetup.AuctionVariables.SlotDeadline)
391 stats.Sync.LastBlock = stats.Eth.LastBlock
392 stats.Sync.Auction.CurrentSlot.Forger = bootForger
393 assert.Equal(t, true, coord.canForge(stats))
394 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
395 assert.NotNil(t, coord.pipeline)
396
397 // Slot 0. No bid, and we reach the deadline, so anyone can forge
398 // pipelineStarted: true -> true
399 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
400 int64(ethClientSetup.AuctionVariables.SlotDeadline) + 1
401 stats.Sync.LastBlock = stats.Eth.LastBlock
402 stats.Sync.Auction.CurrentSlot.Forger = bootForger
403 assert.Equal(t, true, coord.canForge(stats))
404 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
405 assert.NotNil(t, coord.pipeline)
406
407 // Slot 0. No bid, so the winner is the boot coordinator
408 // pipelineStarted: true -> false
409 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
410 1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
411 stats.Sync.LastBlock = stats.Eth.LastBlock
412 stats.Sync.Auction.CurrentSlot.Forger = bootForger
413 assert.Equal(t, false, coord.canForge(stats))
414 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
415 assert.Nil(t, coord.pipeline)
416}
417
418func TestPipelineShouldL1L2Batch(t *testing.T) {
419 ethClientSetup := test.NewClientSetupExample()
420
421 var timer timer
422 ctx := context.Background()
423 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
424 modules := newTestModules(t)
425 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
426 pipeline, err := coord.newPipeline(ctx)
427 require.NoError(t, err)
428 pipeline.vars = coord.vars
429
430 // Check that the parameters are the ones we expect and use in this test
431 require.Equal(t, 0.5, pipeline.cfg.L1BatchTimeoutPerc)
432 require.Equal(t, int64(9), ethClientSetup.RollupVariables.ForgeL1L2BatchTimeout)
433 l1BatchTimeoutPerc := pipeline.cfg.L1BatchTimeoutPerc
434 l1BatchTimeout := ethClientSetup.RollupVariables.ForgeL1L2BatchTimeout
435
436 var stats synchronizer.Stats
437
438 startBlock := int64(100)
439
440 //
441 // No scheduled L1Batch
442 //
443
444 // Last L1Batch was a long time ago
445 stats.Eth.LastBlock.Num = startBlock
446 stats.Sync.LastBlock = stats.Eth.LastBlock
447 stats.Sync.LastL1BatchBlock = 0
448 pipeline.stats = stats
449 assert.Equal(t, true, pipeline.shouldL1L2Batch())
450
451 stats.Sync.LastL1BatchBlock = startBlock
452
453 // We are are one block before the timeout range * 0.5
454 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc) - 1
455 stats.Sync.LastBlock = stats.Eth.LastBlock
456 pipeline.stats = stats
457 assert.Equal(t, false, pipeline.shouldL1L2Batch())
458
459 // We are are at timeout range * 0.5
460 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc)
461 stats.Sync.LastBlock = stats.Eth.LastBlock
462 pipeline.stats = stats
463 assert.Equal(t, true, pipeline.shouldL1L2Batch())
464
465 //
466 // Scheduled L1Batch
467 //
468 pipeline.lastScheduledL1BatchBlockNum = startBlock
469 stats.Sync.LastL1BatchBlock = startBlock - 10
470
471 // We are are one block before the timeout range * 0.5
472 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc) - 1
473 stats.Sync.LastBlock = stats.Eth.LastBlock
474 pipeline.stats = stats
475 assert.Equal(t, false, pipeline.shouldL1L2Batch())
476
477 // We are are at timeout range * 0.5
478 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc)
479 stats.Sync.LastBlock = stats.Eth.LastBlock
480 pipeline.stats = stats
481 assert.Equal(t, true, pipeline.shouldL1L2Batch())
482}
483
484// ethAddTokens adds the tokens from the blocks to the blockchain
485func ethAddTokens(blocks []common.BlockData, client *test.Client) {
486 for _, block := range blocks {
487 for _, token := range block.Rollup.AddedTokens {
488 consts := eth.ERC20Consts{
489 Name: fmt.Sprintf("Token %d", token.TokenID),
490 Symbol: fmt.Sprintf("TK%d", token.TokenID),
491 Decimals: 18,
492 }
493 // tokenConsts[token.TokenID] = consts
494 client.CtlAddERC20(token.EthAddr, consts)
495 }
496 }
497}
498
499const testTokensLen = 3
500const testUsersLen = 4
501
502func preloadSync(t *testing.T, ethClient *test.Client, sync *synchronizer.Synchronizer,
503 historyDB *historydb.HistoryDB, stateDB *statedb.StateDB) *til.Context {
504 // Create a set with `testTokensLen` tokens and for each token
505 // `testUsersLen` accounts.
506 var set []til.Instruction
507 // set = append(set, til.Instruction{Typ: "Blockchain"})
508 for tokenID := 1; tokenID < testTokensLen; tokenID++ {
509 set = append(set, til.Instruction{
510 Typ: til.TypeAddToken,
511 TokenID: common.TokenID(tokenID),
512 })
513 }
514 depositAmount, ok := new(big.Int).SetString("10225000000000000000000000000000000", 10)
515 require.True(t, ok)
516 for tokenID := 0; tokenID < testTokensLen; tokenID++ {
517 for user := 0; user < testUsersLen; user++ {
518 set = append(set, til.Instruction{
519 Typ: common.TxTypeCreateAccountDeposit,
520 TokenID: common.TokenID(tokenID),
521 DepositAmount: depositAmount,
522 From: fmt.Sprintf("User%d", user),
523 })
524 }
525 }
526 set = append(set, til.Instruction{Typ: til.TypeNewBatchL1})
527 set = append(set, til.Instruction{Typ: til.TypeNewBatchL1})
528 set = append(set, til.Instruction{Typ: til.TypeNewBlock})
529
530 tc := til.NewContext(chainID, common.RollupConstMaxL1UserTx)
531 blocks, err := tc.GenerateBlocksFromInstructions(set)
532 require.NoError(t, err)
533 require.NotNil(t, blocks)
534
535 ethAddTokens(blocks, ethClient)
536 err = ethClient.CtlAddBlocks(blocks)
537 require.NoError(t, err)
538
539 ctx := context.Background()
540 for {
541 syncBlock, discards, err := sync.Sync2(ctx, nil)
542 require.NoError(t, err)
543 require.Nil(t, discards)
544 if syncBlock == nil {
545 break
546 }
547 }
548 dbTokens, err := historyDB.GetAllTokens()
549 require.Nil(t, err)
550 require.Equal(t, testTokensLen, len(dbTokens))
551
552 dbAccounts, err := historyDB.GetAllAccounts()
553 require.Nil(t, err)
554 require.Equal(t, testTokensLen*testUsersLen, len(dbAccounts))
555
556 sdbAccounts, err := stateDB.GetAccounts()
557 require.Nil(t, err)
558 require.Equal(t, testTokensLen*testUsersLen, len(sdbAccounts))
559
560 return tc
561}
562
563func TestPipeline1(t *testing.T) {
564 ethClientSetup := test.NewClientSetupExample()
565
566 var timer timer
567 ctx := context.Background()
568 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
569 modules := newTestModules(t)
570 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
571 sync := newTestSynchronizer(t, ethClient, ethClientSetup, modules)
572 pipeline, err := coord.newPipeline(ctx)
573 require.NoError(t, err)
574
575 require.NotNil(t, sync)
576 require.NotNil(t, pipeline)
577
578 // preload the synchronier (via the test ethClient) some tokens and
579 // users with positive balances
580 tilCtx := preloadSync(t, ethClient, sync, modules.historyDB, modules.stateDB)
581 syncStats := sync.Stats()
582 batchNum := common.BatchNum(syncStats.Sync.LastBatch)
583 syncSCVars := sync.SCVars()
584
585 // Insert some l2txs in the Pool
586 setPool := `
587Type: PoolL2
588
589PoolTransfer(0) User0-User1: 100 (126)
590PoolTransfer(0) User1-User2: 200 (126)
591PoolTransfer(0) User2-User3: 300 (126)
592 `
593 l2txs, err := tilCtx.GeneratePoolL2Txs(setPool)
594 require.NoError(t, err)
595 for _, tx := range l2txs {
596 err := modules.l2DB.AddTxTest(&tx) //nolint:gosec
597 require.NoError(t, err)
598 }
599
600 err = pipeline.reset(batchNum, syncStats.Sync.LastForgeL1TxsNum, &synchronizer.SCVariables{
601 Rollup: *syncSCVars.Rollup,
602 Auction: *syncSCVars.Auction,
603 WDelayer: *syncSCVars.WDelayer,
604 })
605 require.NoError(t, err)
606 // Sanity check
607 sdbAccounts, err := pipeline.txSelector.LocalAccountsDB().GetAccounts()
608 require.Nil(t, err)
609 require.Equal(t, testTokensLen*testUsersLen, len(sdbAccounts))
610
611 // Sanity check
612 sdbAccounts, err = pipeline.batchBuilder.LocalStateDB().GetAccounts()
613 require.Nil(t, err)
614 require.Equal(t, testTokensLen*testUsersLen, len(sdbAccounts))
615
616 // Sanity check
617 require.Equal(t, modules.stateDB.MerkleTree().Root(),
618 pipeline.batchBuilder.LocalStateDB().MerkleTree().Root())
619
620 batchNum++
621
622 selectionConfig := &txselector.SelectionConfig{
623 MaxL1UserTxs: maxL1UserTxs,
624 MaxL1CoordinatorTxs: maxL1CoordinatorTxs,
625 ProcessTxsConfig: statedb.ProcessTxsConfig{
626 NLevels: nLevels,
627 MaxFeeTx: maxFeeTxs,
628 MaxTx: uint32(maxTxs),
629 MaxL1Tx: uint32(maxL1Txs),
630 },
631 }
632
633 batchInfo, err := pipeline.forgeBatch(ctx, batchNum, selectionConfig)
634 require.NoError(t, err)
635 assert.Equal(t, 3, len(batchInfo.L2Txs))
636
637 batchNum++
638 batchInfo, err = pipeline.forgeBatch(ctx, batchNum, selectionConfig)
639 require.NoError(t, err)
640 assert.Equal(t, 0, len(batchInfo.L2Txs))
641}
642
643func TestCoordinatorStress(t *testing.T) {
644 if os.Getenv("TEST_COORD_STRESS") == "" {
645 return
646 }
647 log.Info("Begin Test Coord Stress")
648 ethClientSetup := test.NewClientSetupExample()
649 var timer timer
650 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
651 modules := newTestModules(t)
652 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
653 syn := newTestSynchronizer(t, ethClient, ethClientSetup, modules)
654
655 coord.Start()
656 ctx, cancel := context.WithCancel(context.Background())
657 var wg sync.WaitGroup
658
659 // Synchronizer loop
660 wg.Add(1)
661 go func() {
662 for {
663 blockData, _, err := syn.Sync2(ctx, nil)
664 if ctx.Err() != nil {
665 wg.Done()
666 return
667 }
668 require.NoError(t, err)
669 if blockData != nil {
670 stats := syn.Stats()
671 coord.SendMsg(MsgSyncBlock{
672 Stats: *stats,
673 Batches: blockData.Rollup.Batches,
674 Vars: synchronizer.SCVariablesPtr{
675 Rollup: blockData.Rollup.Vars,
676 Auction: blockData.Auction.Vars,
677 WDelayer: blockData.WDelayer.Vars,
678 },
679 })
680 } else {
681 time.Sleep(100 * time.Millisecond)
682 }
683 }
684 }()
685
686 // Blockchain mining loop
687 wg.Add(1)
688 go func() {
689 for {
690 select {
691 case <-ctx.Done():
692 wg.Done()
693 return
694 case <-time.After(100 * time.Millisecond):
695 ethClient.CtlMineBlock()
696 }
697 }
698 }()
699
700 time.Sleep(600 * time.Second)
701
702 cancel()
703 wg.Wait()
704 coord.Stop()
705}
706
707// TODO: Test Reorg
708// TODO: Test Pipeline
709// TODO: Test TxMonitor
710// TODO: Test forgeBatch
711// TODO: Test waitServerProof
712// TODO: Test handleReorg