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 ethClientSetup.ChainID = big.NewInt(int64(chainID))
217 var timer timer
218 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
219 modules := newTestModules(t)
220 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
221
222 // Bid for slot 2 and 4
223 _, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
224 require.NoError(t, err)
225 _, err = ethClient.AuctionBidSimple(2, big.NewInt(9999))
226 require.NoError(t, err)
227 _, err = ethClient.AuctionBidSimple(4, big.NewInt(9999))
228 require.NoError(t, err)
229
230 coord.Start()
231 time.Sleep(1 * time.Second)
232
233 waitForSlot := func(slot int64) {
234 for {
235 blockNum, err := ethClient.EthLastBlock()
236 require.NoError(t, err)
237 nextBlockSlot, err := ethClient.AuctionGetSlotNumber(blockNum + 1)
238 require.NoError(t, err)
239 if nextBlockSlot == slot {
240 break
241 }
242 ethClient.CtlMineBlock()
243 time.Sleep(100 * time.Millisecond)
244 var stats synchronizer.Stats
245 stats.Eth.LastBlock = *ethClient.CtlLastBlock()
246 stats.Sync.LastBlock = stats.Eth.LastBlock
247 stats.Eth.LastBatch = ethClient.CtlLastForgedBatch()
248 stats.Sync.LastBatch = stats.Eth.LastBatch
249 canForge, err := ethClient.AuctionCanForge(forger, blockNum+1)
250 require.NoError(t, err)
251 if canForge {
252 // fmt.Println("DBG canForge")
253 stats.Sync.Auction.CurrentSlot.Forger = forger
254 }
255 // Copy stateDB to synchronizer if there was a new batch
256 source := fmt.Sprintf("%v/BatchNum%v", batchBuilderDBPath, stats.Sync.LastBatch)
257 dest := fmt.Sprintf("%v/BatchNum%v", syncDBPath, stats.Sync.LastBatch)
258 if stats.Sync.LastBatch != 0 {
259 if _, err := os.Stat(dest); os.IsNotExist(err) {
260 log.Infow("Making pebble checkpoint for sync",
261 "source", source, "dest", dest)
262 err = pebbleMakeCheckpoint(source, dest)
263 require.NoError(t, err)
264 }
265 }
266 coord.SendMsg(MsgSyncBlock{
267 Stats: stats,
268 })
269 }
270 }
271
272 // NOTE: With the current test, the coordinator will enter in forge
273 // time before the bidded slot because no one else is forging in the
274 // other slots before the slot deadline.
275 // simulate forgeSequence time
276 waitForSlot(2)
277 log.Info("~~~ simulate entering in forge time")
278 time.Sleep(1 * time.Second)
279
280 // simulate going out from forgeSequence
281 waitForSlot(3)
282 log.Info("~~~ simulate going out from forge time")
283 time.Sleep(1 * time.Second)
284
285 // simulate entering forgeSequence time again
286 waitForSlot(4)
287 log.Info("~~~ simulate entering in forge time again")
288 time.Sleep(2 * time.Second)
289
290 // simulate stopping forgerLoop by channel
291 log.Info("~~~ simulate stopping forgerLoop by closing coordinator stopch")
292 coord.Stop()
293 time.Sleep(1 * time.Second)
294}
295
296func TestCoordinatorStartStop(t *testing.T) {
297 ethClientSetup := test.NewClientSetupExample()
298 ethClientSetup.ChainID = big.NewInt(int64(chainID))
299 var timer timer
300 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
301 modules := newTestModules(t)
302 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
303 coord.Start()
304 coord.Stop()
305}
306
307func TestCoordCanForge(t *testing.T) {
308 ethClientSetup := test.NewClientSetupExample()
309 ethClientSetup.ChainID = big.NewInt(int64(chainID))
310 bootForger := ethClientSetup.AuctionVariables.BootCoordinator
311
312 var timer timer
313 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
314 modules := newTestModules(t)
315 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
316 _, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
317 require.NoError(t, err)
318 bid, ok := new(big.Int).SetString("12000000000000000000", 10)
319 if !ok {
320 panic("bad bid")
321 }
322 _, err = ethClient.AuctionBidSimple(2, bid)
323 require.NoError(t, err)
324
325 modules2 := newTestModules(t)
326 bootCoord := newTestCoordinator(t, bootForger, ethClient, ethClientSetup, modules2)
327
328 assert.Equal(t, forger, coord.cfg.ForgerAddress)
329 assert.Equal(t, bootForger, bootCoord.cfg.ForgerAddress)
330 ethBootCoord, err := ethClient.AuctionGetBootCoordinator()
331 require.NoError(t, err)
332 assert.Equal(t, &bootForger, ethBootCoord)
333
334 var stats synchronizer.Stats
335
336 // Slot 0. No bid, so the winner is the boot coordinator
337 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
338 stats.Sync.LastBlock = stats.Eth.LastBlock
339 stats.Sync.Auction.CurrentSlot.Forger = bootForger
340 assert.Equal(t, false, coord.canForge(&stats))
341 assert.Equal(t, true, bootCoord.canForge(&stats))
342
343 // Slot 0. No bid, and we reach the deadline, so anyone can forge
344 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
345 int64(ethClientSetup.AuctionVariables.SlotDeadline)
346 stats.Sync.LastBlock = stats.Eth.LastBlock
347 stats.Sync.Auction.CurrentSlot.Forger = bootForger
348 assert.Equal(t, true, coord.canForge(&stats))
349 assert.Equal(t, true, bootCoord.canForge(&stats))
350
351 // Slot 1. coordinator bid, so the winner is the coordinator
352 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
353 1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
354 stats.Sync.LastBlock = stats.Eth.LastBlock
355 stats.Sync.Auction.CurrentSlot.Forger = forger
356 assert.Equal(t, true, coord.canForge(&stats))
357 assert.Equal(t, false, bootCoord.canForge(&stats))
358}
359
360func TestCoordHandleMsgSyncBlock(t *testing.T) {
361 ethClientSetup := test.NewClientSetupExample()
362 ethClientSetup.ChainID = big.NewInt(int64(chainID))
363 bootForger := ethClientSetup.AuctionVariables.BootCoordinator
364
365 var timer timer
366 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
367 modules := newTestModules(t)
368 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
369 _, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
370 require.NoError(t, err)
371 bid, ok := new(big.Int).SetString("11000000000000000000", 10)
372 if !ok {
373 panic("bad bid")
374 }
375 _, err = ethClient.AuctionBidSimple(2, bid)
376 require.NoError(t, err)
377
378 var msg MsgSyncBlock
379 stats := &msg.Stats
380 ctx := context.Background()
381
382 // Slot 0. No bid, so the winner is the boot coordinator
383 // pipelineStarted: false -> false
384 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
385 stats.Sync.LastBlock = stats.Eth.LastBlock
386 stats.Sync.Auction.CurrentSlot.Forger = bootForger
387 assert.Equal(t, false, coord.canForge(stats))
388 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
389 assert.Nil(t, coord.pipeline)
390
391 // Slot 0. No bid, and we reach the deadline, so anyone can forge
392 // pipelineStarted: false -> true
393 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
394 int64(ethClientSetup.AuctionVariables.SlotDeadline)
395 stats.Sync.LastBlock = stats.Eth.LastBlock
396 stats.Sync.Auction.CurrentSlot.Forger = bootForger
397 assert.Equal(t, true, coord.canForge(stats))
398 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
399 assert.NotNil(t, coord.pipeline)
400
401 // Slot 0. No bid, and we reach the deadline, so anyone can forge
402 // pipelineStarted: true -> true
403 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
404 int64(ethClientSetup.AuctionVariables.SlotDeadline) + 1
405 stats.Sync.LastBlock = stats.Eth.LastBlock
406 stats.Sync.Auction.CurrentSlot.Forger = bootForger
407 assert.Equal(t, true, coord.canForge(stats))
408 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
409 assert.NotNil(t, coord.pipeline)
410
411 // Slot 0. No bid, so the winner is the boot coordinator
412 // pipelineStarted: true -> false
413 stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
414 1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
415 stats.Sync.LastBlock = stats.Eth.LastBlock
416 stats.Sync.Auction.CurrentSlot.Forger = bootForger
417 assert.Equal(t, false, coord.canForge(stats))
418 require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
419 assert.Nil(t, coord.pipeline)
420}
421
422func TestPipelineShouldL1L2Batch(t *testing.T) {
423 ethClientSetup := test.NewClientSetupExample()
424 ethClientSetup.ChainID = big.NewInt(int64(chainID))
425
426 var timer timer
427 ctx := context.Background()
428 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
429 modules := newTestModules(t)
430 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
431 pipeline, err := coord.newPipeline(ctx)
432 require.NoError(t, err)
433 pipeline.vars = coord.vars
434
435 // Check that the parameters are the ones we expect and use in this test
436 require.Equal(t, 0.5, pipeline.cfg.L1BatchTimeoutPerc)
437 require.Equal(t, int64(9), ethClientSetup.RollupVariables.ForgeL1L2BatchTimeout)
438 l1BatchTimeoutPerc := pipeline.cfg.L1BatchTimeoutPerc
439 l1BatchTimeout := ethClientSetup.RollupVariables.ForgeL1L2BatchTimeout
440
441 var stats synchronizer.Stats
442
443 startBlock := int64(100)
444
445 //
446 // No scheduled L1Batch
447 //
448
449 // Last L1Batch was a long time ago
450 stats.Eth.LastBlock.Num = startBlock
451 stats.Sync.LastBlock = stats.Eth.LastBlock
452 stats.Sync.LastL1BatchBlock = 0
453 pipeline.stats = stats
454 assert.Equal(t, true, pipeline.shouldL1L2Batch())
455
456 stats.Sync.LastL1BatchBlock = startBlock
457
458 // We are are one block before the timeout range * 0.5
459 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc) - 1
460 stats.Sync.LastBlock = stats.Eth.LastBlock
461 pipeline.stats = stats
462 assert.Equal(t, false, pipeline.shouldL1L2Batch())
463
464 // We are are at timeout range * 0.5
465 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc)
466 stats.Sync.LastBlock = stats.Eth.LastBlock
467 pipeline.stats = stats
468 assert.Equal(t, true, pipeline.shouldL1L2Batch())
469
470 //
471 // Scheduled L1Batch
472 //
473 pipeline.lastScheduledL1BatchBlockNum = startBlock
474 stats.Sync.LastL1BatchBlock = startBlock - 10
475
476 // We are are one block before the timeout range * 0.5
477 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc) - 1
478 stats.Sync.LastBlock = stats.Eth.LastBlock
479 pipeline.stats = stats
480 assert.Equal(t, false, pipeline.shouldL1L2Batch())
481
482 // We are are at timeout range * 0.5
483 stats.Eth.LastBlock.Num = startBlock + int64(float64(l1BatchTimeout)*l1BatchTimeoutPerc)
484 stats.Sync.LastBlock = stats.Eth.LastBlock
485 pipeline.stats = stats
486 assert.Equal(t, true, pipeline.shouldL1L2Batch())
487}
488
489// ethAddTokens adds the tokens from the blocks to the blockchain
490func ethAddTokens(blocks []common.BlockData, client *test.Client) {
491 for _, block := range blocks {
492 for _, token := range block.Rollup.AddedTokens {
493 consts := eth.ERC20Consts{
494 Name: fmt.Sprintf("Token %d", token.TokenID),
495 Symbol: fmt.Sprintf("TK%d", token.TokenID),
496 Decimals: 18,
497 }
498 // tokenConsts[token.TokenID] = consts
499 client.CtlAddERC20(token.EthAddr, consts)
500 }
501 }
502}
503
504const testTokensLen = 3
505const testUsersLen = 4
506
507func preloadSync(t *testing.T, ethClient *test.Client, sync *synchronizer.Synchronizer,
508 historyDB *historydb.HistoryDB, stateDB *statedb.StateDB) *til.Context {
509 // Create a set with `testTokensLen` tokens and for each token
510 // `testUsersLen` accounts.
511 var set []til.Instruction
512 // set = append(set, til.Instruction{Typ: "Blockchain"})
513 for tokenID := 1; tokenID < testTokensLen; tokenID++ {
514 set = append(set, til.Instruction{
515 Typ: til.TypeAddToken,
516 TokenID: common.TokenID(tokenID),
517 })
518 }
519 depositAmount, ok := new(big.Int).SetString("10225000000000000000000000000000000", 10)
520 require.True(t, ok)
521 for tokenID := 0; tokenID < testTokensLen; tokenID++ {
522 for user := 0; user < testUsersLen; user++ {
523 set = append(set, til.Instruction{
524 Typ: common.TxTypeCreateAccountDeposit,
525 TokenID: common.TokenID(tokenID),
526 DepositAmount: depositAmount,
527 From: fmt.Sprintf("User%d", user),
528 })
529 }
530 }
531 set = append(set, til.Instruction{Typ: til.TypeNewBatchL1})
532 set = append(set, til.Instruction{Typ: til.TypeNewBatchL1})
533 set = append(set, til.Instruction{Typ: til.TypeNewBlock})
534
535 tc := til.NewContext(chainID, common.RollupConstMaxL1UserTx)
536 blocks, err := tc.GenerateBlocksFromInstructions(set)
537 require.NoError(t, err)
538 require.NotNil(t, blocks)
539
540 ethAddTokens(blocks, ethClient)
541 err = ethClient.CtlAddBlocks(blocks)
542 require.NoError(t, err)
543
544 ctx := context.Background()
545 for {
546 syncBlock, discards, err := sync.Sync2(ctx, nil)
547 require.NoError(t, err)
548 require.Nil(t, discards)
549 if syncBlock == nil {
550 break
551 }
552 }
553 dbTokens, err := historyDB.GetAllTokens()
554 require.Nil(t, err)
555 require.Equal(t, testTokensLen, len(dbTokens))
556
557 dbAccounts, err := historyDB.GetAllAccounts()
558 require.Nil(t, err)
559 require.Equal(t, testTokensLen*testUsersLen, len(dbAccounts))
560
561 sdbAccounts, err := stateDB.GetAccounts()
562 require.Nil(t, err)
563 require.Equal(t, testTokensLen*testUsersLen, len(sdbAccounts))
564
565 return tc
566}
567
568func TestPipeline1(t *testing.T) {
569 ethClientSetup := test.NewClientSetupExample()
570 ethClientSetup.ChainID = big.NewInt(int64(chainID))
571
572 var timer timer
573 ctx := context.Background()
574 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
575 modules := newTestModules(t)
576 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
577 sync := newTestSynchronizer(t, ethClient, ethClientSetup, modules)
578 pipeline, err := coord.newPipeline(ctx)
579 require.NoError(t, err)
580
581 require.NotNil(t, sync)
582 require.NotNil(t, pipeline)
583
584 // preload the synchronier (via the test ethClient) some tokens and
585 // users with positive balances
586 tilCtx := preloadSync(t, ethClient, sync, modules.historyDB, modules.stateDB)
587 syncStats := sync.Stats()
588 batchNum := common.BatchNum(syncStats.Sync.LastBatch)
589 syncSCVars := sync.SCVars()
590
591 // Insert some l2txs in the Pool
592 setPool := `
593Type: PoolL2
594
595PoolTransfer(0) User0-User1: 100 (126)
596PoolTransfer(0) User1-User2: 200 (126)
597PoolTransfer(0) User2-User3: 300 (126)
598 `
599 l2txs, err := tilCtx.GeneratePoolL2Txs(setPool)
600 require.NoError(t, err)
601 for _, tx := range l2txs {
602 err := modules.l2DB.AddTxTest(&tx) //nolint:gosec
603 require.NoError(t, err)
604 }
605
606 err = pipeline.reset(batchNum, syncStats.Sync.LastForgeL1TxsNum, &synchronizer.SCVariables{
607 Rollup: *syncSCVars.Rollup,
608 Auction: *syncSCVars.Auction,
609 WDelayer: *syncSCVars.WDelayer,
610 })
611 require.NoError(t, err)
612 // Sanity check
613 sdbAccounts, err := pipeline.txSelector.LocalAccountsDB().GetAccounts()
614 require.Nil(t, err)
615 require.Equal(t, testTokensLen*testUsersLen, len(sdbAccounts))
616
617 // Sanity check
618 sdbAccounts, err = pipeline.batchBuilder.LocalStateDB().GetAccounts()
619 require.Nil(t, err)
620 require.Equal(t, testTokensLen*testUsersLen, len(sdbAccounts))
621
622 // Sanity check
623 require.Equal(t, modules.stateDB.MerkleTree().Root(),
624 pipeline.batchBuilder.LocalStateDB().MerkleTree().Root())
625
626 batchNum++
627
628 selectionConfig := &txselector.SelectionConfig{
629 MaxL1UserTxs: maxL1UserTxs,
630 MaxL1CoordinatorTxs: maxL1CoordinatorTxs,
631 ProcessTxsConfig: statedb.ProcessTxsConfig{
632 NLevels: nLevels,
633 MaxFeeTx: maxFeeTxs,
634 MaxTx: uint32(maxTxs),
635 MaxL1Tx: uint32(maxL1Txs),
636 },
637 }
638
639 batchInfo, err := pipeline.forgeBatch(ctx, batchNum, selectionConfig)
640 require.NoError(t, err)
641 assert.Equal(t, 3, len(batchInfo.L2Txs))
642
643 batchNum++
644 batchInfo, err = pipeline.forgeBatch(ctx, batchNum, selectionConfig)
645 require.NoError(t, err)
646 assert.Equal(t, 0, len(batchInfo.L2Txs))
647}
648
649func TestCoordinatorStress(t *testing.T) {
650 if os.Getenv("TEST_COORD_STRESS") == "" {
651 return
652 }
653 log.Info("Begin Test Coord Stress")
654 ethClientSetup := test.NewClientSetupExample()
655 ethClientSetup.ChainID = big.NewInt(int64(chainID))
656 var timer timer
657 ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
658 modules := newTestModules(t)
659 coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
660 syn := newTestSynchronizer(t, ethClient, ethClientSetup, modules)
661
662 coord.Start()
663 ctx, cancel := context.WithCancel(context.Background())
664 var wg sync.WaitGroup
665
666 // Synchronizer loop
667 wg.Add(1)
668 go func() {
669 for {
670 blockData, _, err := syn.Sync2(ctx, nil)
671 if ctx.Err() != nil {
672 wg.Done()
673 return
674 }
675 require.NoError(t, err)
676 if blockData != nil {
677 stats := syn.Stats()
678 coord.SendMsg(MsgSyncBlock{
679 Stats: *stats,
680 Batches: blockData.Rollup.Batches,
681 Vars: synchronizer.SCVariablesPtr{
682 Rollup: blockData.Rollup.Vars,
683 Auction: blockData.Auction.Vars,
684 WDelayer: blockData.WDelayer.Vars,
685 },
686 })
687 } else {
688 time.Sleep(100 * time.Millisecond)
689 }
690 }
691 }()
692
693 // Blockchain mining loop
694 wg.Add(1)
695 go func() {
696 for {
697 select {
698 case <-ctx.Done():
699 wg.Done()
700 return
701 case <-time.After(1 * time.Second):
702 ethClient.CtlMineBlock()
703 }
704 }
705 }()
706
707 time.Sleep(600 * time.Second)
708
709 cancel()
710 wg.Wait()
711 coord.Stop()
712}
713
714// TODO: Test Reorg
715// TODO: Test Pipeline
716// TODO: Test TxMonitor
717// TODO: Test forgeBatch
718// TODO: Test waitServerProof
719// TODO: Test handleReorg