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