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