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/txprocessor"
 26	"github.com/hermeznetwork/hermez-node/txselector"
 27	"github.com/hermeznetwork/tracerr"
 28	"github.com/iden3/go-iden3-crypto/babyjub"
 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 maxL1Txs uint64 = 256
 90var maxTxs uint64 = 376
 91var nLevels uint32 = 32   //nolint:deadcode,unused
 92var maxFeeTxs uint32 = 64 //nolint:deadcode,varcheck
 93var chainID uint16 = 0
 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, 128, 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	var bjj babyjub.PublicKeyComp
115	err = bjj.UnmarshalText([]byte("c433f7a696b7aa3a5224efb3993baf0ccd9e92eecee0c29a3f6c8208a9e81d9e"))
116	require.NoError(t, err)
117	coordAccount := &txselector.CoordAccount{ // TODO TMP
118		Addr:                ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
119		BJJ:                 bjj,
120		AccountCreationAuth: nil,
121	}
122	txSelector, err := txselector.NewTxSelector(coordAccount, txSelDBPath, syncStateDB, l2DB)
123	assert.NoError(t, err)
124
125	batchBuilderDBPath, err = ioutil.TempDir("", "tmpBatchBuilderDB")
126	require.NoError(t, err)
127	deleteme = append(deleteme, batchBuilderDBPath)
128	batchBuilder, err := batchbuilder.NewBatchBuilder(batchBuilderDBPath, syncStateDB, 0, uint64(nLevels))
129	assert.NoError(t, err)
130
131	return modules{
132		historyDB:    historyDB,
133		l2DB:         l2DB,
134		txSelector:   txSelector,
135		batchBuilder: batchBuilder,
136		stateDB:      syncStateDB,
137	}
138}
139
140type timer struct {
141	time int64
142}
143
144func (t *timer) Time() int64 {
145	currentTime := t.time
146	t.time++
147	return currentTime
148}
149
150var bidder = ethCommon.HexToAddress("0x6b175474e89094c44da98b954eedeac495271d0f")
151var forger = ethCommon.HexToAddress("0xc344E203a046Da13b0B4467EB7B3629D0C99F6E6")
152
153func newTestCoordinator(t *testing.T, forgerAddr ethCommon.Address, ethClient *test.Client,
154	ethClientSetup *test.ClientSetup, modules modules) *Coordinator {
155	debugBatchPath, err := ioutil.TempDir("", "tmpDebugBatch")
156	require.NoError(t, err)
157	deleteme = append(deleteme, debugBatchPath)
158
159	conf := Config{
160		ForgerAddress:          forgerAddr,
161		ConfirmBlocks:          5,
162		L1BatchTimeoutPerc:     0.5,
163		EthClientAttempts:      5,
164		SyncRetryInterval:      400 * time.Microsecond,
165		EthClientAttemptsDelay: 100 * time.Millisecond,
166		TxManagerCheckInterval: 300 * time.Millisecond,
167		DebugBatchPath:         debugBatchPath,
168		Purger: PurgerCfg{
169			PurgeBatchDelay:      10,
170			PurgeBlockDelay:      10,
171			InvalidateBatchDelay: 4,
172			InvalidateBlockDelay: 4,
173		},
174		TxProcessorConfig: txprocessor.Config{
175			NLevels:  nLevels,
176			MaxFeeTx: maxFeeTxs,
177			MaxTx:    uint32(maxTxs),
178			MaxL1Tx:  uint32(maxL1Txs),
179			ChainID:  chainID,
180		},
181		VerifierIdx: 0,
182	}
183
184	serverProofs := []prover.Client{
185		&prover.MockClient{Delay: 300 * time.Millisecond},
186		&prover.MockClient{Delay: 400 * time.Millisecond},
187	}
188
189	scConsts := &synchronizer.SCConsts{
190		Rollup:   *ethClientSetup.RollupConstants,
191		Auction:  *ethClientSetup.AuctionConstants,
192		WDelayer: *ethClientSetup.WDelayerConstants,
193	}
194	initSCVars := &synchronizer.SCVariables{
195		Rollup:   *ethClientSetup.RollupVariables,
196		Auction:  *ethClientSetup.AuctionVariables,
197		WDelayer: *ethClientSetup.WDelayerVariables,
198	}
199	coord, err := NewCoordinator(conf, modules.historyDB, modules.l2DB, modules.txSelector,
200		modules.batchBuilder, serverProofs, ethClient, scConsts, initSCVars)
201	require.NoError(t, err)
202	return coord
203}
204
205func newTestSynchronizer(t *testing.T, ethClient *test.Client, ethClientSetup *test.ClientSetup,
206	modules modules) *synchronizer.Synchronizer {
207	sync, err := synchronizer.NewSynchronizer(ethClient, modules.historyDB, modules.stateDB,
208		synchronizer.Config{
209			StatsRefreshPeriod: 0 * time.Second,
210		})
211	require.NoError(t, err)
212	return sync
213}
214
215// TestCoordinatorFlow is a test where the coordinator is stared (which means
216// that goroutines are spawned), and ethereum blocks are mined via the
217// test.Client to simulate starting and stopping forging times.  This test
218// works without a synchronizer, and no l2txs are inserted in the pool, so all
219// the batches are forged empty.  The purpose of this test is to manually
220// observe via the logs that nothing crashes and that the coordinator starts
221// and stops forging at the right blocks.
222func TestCoordinatorFlow(t *testing.T) {
223	if os.Getenv("TEST_COORD_FLOW") == "" {
224		return
225	}
226	ethClientSetup := test.NewClientSetupExample()
227	ethClientSetup.ChainID = big.NewInt(int64(chainID))
228	var timer timer
229	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
230	modules := newTestModules(t)
231	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
232
233	// Bid for slot 2 and 4
234	_, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
235	require.NoError(t, err)
236	bid, ok := new(big.Int).SetString("11000000000000000000", 10)
237	if !ok {
238		panic("bad bid")
239	}
240	_, err = ethClient.AuctionBidSimple(3, bid)
241	require.NoError(t, err)
242	_, err = ethClient.AuctionBidSimple(5, bid)
243	require.NoError(t, err)
244
245	coord.Start()
246	ctx := context.Background()
247	time.Sleep(1 * time.Second)
248
249	waitForSlot := func(slot int64) {
250		for {
251			blockNum, err := ethClient.EthLastBlock()
252			require.NoError(t, err)
253			nextBlockSlot, err := ethClient.AuctionGetSlotNumber(blockNum + 1)
254			require.NoError(t, err)
255			if nextBlockSlot == slot {
256				break
257			}
258			ethClient.CtlMineBlock()
259			time.Sleep(100 * time.Millisecond)
260			var stats synchronizer.Stats
261			stats.Eth.LastBlock = *ethClient.CtlLastBlock()
262			stats.Sync.LastBlock = stats.Eth.LastBlock
263			stats.Eth.LastBatch = ethClient.CtlLastForgedBatch()
264			stats.Sync.LastBatch = stats.Eth.LastBatch
265			canForge, err := ethClient.AuctionCanForge(forger, blockNum+1)
266			require.NoError(t, err)
267			var slot common.Slot
268			slotNum := ethClientSetup.AuctionConstants.SlotNum(blockNum + 1)
269			slot.StartBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
270				(slotNum)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
271			slot.EndBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
272				(slotNum+1)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
273			if canForge {
274				slot.Forger = forger
275			}
276			stats.Sync.Auction.CurrentSlot = slot
277
278			// Copy stateDB to synchronizer if there was a new batch
279			source := fmt.Sprintf("%v/BatchNum%v", batchBuilderDBPath, stats.Sync.LastBatch)
280			dest := fmt.Sprintf("%v/BatchNum%v", syncDBPath, stats.Sync.LastBatch)
281			if stats.Sync.LastBatch != 0 {
282				if _, err := os.Stat(dest); os.IsNotExist(err) {
283					log.Infow("Making pebble checkpoint for sync",
284						"source", source, "dest", dest)
285					err = pebbleMakeCheckpoint(source, dest)
286					require.NoError(t, err)
287				}
288			}
289			coord.SendMsg(ctx, MsgSyncBlock{
290				Stats: stats,
291			})
292		}
293	}
294
295	// NOTE: With the current test, the coordinator will enter in forge
296	// time before the bidded slot because no one else is forging in the
297	// other slots before the slot deadline.
298	// simulate forgeSequence time
299	waitForSlot(2)
300	log.Info("~~~ simulate entering in forge time")
301	time.Sleep(1 * time.Second)
302
303	// simulate going out from forgeSequence
304	waitForSlot(3)
305	log.Info("~~~ simulate going out from forge time")
306	time.Sleep(1 * time.Second)
307
308	// simulate entering forgeSequence time again
309	waitForSlot(4)
310	log.Info("~~~ simulate entering in forge time again")
311	time.Sleep(2 * time.Second)
312
313	// simulate stopping forgerLoop by channel
314	log.Info("~~~ simulate stopping forgerLoop by closing coordinator stopch")
315	coord.Stop()
316	time.Sleep(1 * time.Second)
317}
318
319func TestCoordinatorStartStop(t *testing.T) {
320	ethClientSetup := test.NewClientSetupExample()
321	ethClientSetup.ChainID = big.NewInt(int64(chainID))
322	var timer timer
323	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
324	modules := newTestModules(t)
325	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
326	coord.Start()
327	coord.Stop()
328}
329
330func TestCoordCanForge(t *testing.T) {
331	ethClientSetup := test.NewClientSetupExample()
332	ethClientSetup.ChainID = big.NewInt(int64(chainID))
333	bootForger := ethClientSetup.AuctionVariables.BootCoordinator
334
335	var timer timer
336	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
337	modules := newTestModules(t)
338	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
339	_, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
340	require.NoError(t, err)
341	bid, ok := new(big.Int).SetString("12000000000000000000", 10)
342	if !ok {
343		panic("bad bid")
344	}
345	_, err = ethClient.AuctionBidSimple(3, bid)
346	require.NoError(t, err)
347
348	modules2 := newTestModules(t)
349	bootCoord := newTestCoordinator(t, bootForger, ethClient, ethClientSetup, modules2)
350
351	assert.Equal(t, forger, coord.cfg.ForgerAddress)
352	assert.Equal(t, bootForger, bootCoord.cfg.ForgerAddress)
353	ethBootCoord, err := ethClient.AuctionGetBootCoordinator()
354	require.NoError(t, err)
355	assert.Equal(t, &bootForger, ethBootCoord)
356
357	var stats synchronizer.Stats
358
359	slots := [4]common.Slot{}
360	for i := 0; i < 4; i++ {
361		slots[i].StartBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
362			int64(i)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
363		slots[i].EndBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
364			int64(i+1)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
365		// Only slot 3 has Coordinator winner, the rest are BootCoordinator
366		if i == 3 {
367			slots[i].Forger = forger
368		} else {
369			slots[i].Forger = bootForger
370		}
371	}
372
373	// Slot 0.  No bid, so the winner is the boot coordinator
374	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
375	stats.Sync.LastBlock = stats.Eth.LastBlock
376	stats.Sync.Auction.CurrentSlot = slots[0]
377	coord.stats = stats
378	bootCoord.stats = stats
379	assert.Equal(t, false, coord.canForge())
380	assert.Equal(t, true, bootCoord.canForge())
381
382	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
383	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
384		int64(ethClientSetup.AuctionVariables.SlotDeadline)
385	stats.Sync.LastBlock = stats.Eth.LastBlock
386	stats.Sync.Auction.CurrentSlot = slots[0]
387	coord.stats = stats
388	bootCoord.stats = stats
389	assert.Equal(t, true, coord.canForge())
390	assert.Equal(t, true, bootCoord.canForge())
391
392	// Slot 3. coordinator bid, so the winner is the coordinator
393	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
394		3*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
395	stats.Sync.LastBlock = stats.Eth.LastBlock
396	stats.Sync.Auction.CurrentSlot = slots[3]
397	coord.stats = stats
398	bootCoord.stats = stats
399	assert.Equal(t, true, coord.canForge())
400	assert.Equal(t, false, bootCoord.canForge())
401}
402
403func TestCoordHandleMsgSyncBlock(t *testing.T) {
404	ethClientSetup := test.NewClientSetupExample()
405	ethClientSetup.ChainID = big.NewInt(int64(chainID))
406	bootForger := ethClientSetup.AuctionVariables.BootCoordinator
407
408	var timer timer
409	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
410	modules := newTestModules(t)
411	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
412	_, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
413	require.NoError(t, err)
414	bid, ok := new(big.Int).SetString("11000000000000000000", 10)
415	if !ok {
416		panic("bad bid")
417	}
418	_, err = ethClient.AuctionBidSimple(3, bid)
419	require.NoError(t, err)
420
421	slots := [4]common.Slot{}
422	for i := 0; i < 4; i++ {
423		slots[i].StartBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
424			int64(i)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
425		slots[i].EndBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
426			int64(i+1)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
427		// Only slot 3 has Coordinator winner, the rest are BootCoordinator
428		if i == 3 {
429			slots[i].Forger = forger
430		} else {
431			slots[i].Forger = bootForger
432		}
433	}
434
435	var msg MsgSyncBlock
436	coord.stats = msg.Stats
437	ctx := context.Background()
438
439	// Slot 0.  No bid, so the winner is the boot coordinator
440	// pipelineStarted: false -> false
441	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
442	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
443	coord.stats.Sync.Auction.CurrentSlot = slots[0]
444	assert.Equal(t, false, coord.canForge())
445	msg.Stats = coord.stats
446	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
447	assert.Nil(t, coord.pipeline)
448
449	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
450	// pipelineStarted: false -> true
451	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
452		int64(ethClientSetup.AuctionVariables.SlotDeadline)
453	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
454	coord.stats.Sync.Auction.CurrentSlot = slots[0]
455	assert.Equal(t, true, coord.canForge())
456	msg.Stats = coord.stats
457	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
458	assert.NotNil(t, coord.pipeline)
459
460	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
461	// pipelineStarted: true -> true
462	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
463		int64(ethClientSetup.AuctionVariables.SlotDeadline) + 1
464	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
465	coord.stats.Sync.Auction.CurrentSlot = slots[0]
466	assert.Equal(t, true, coord.canForge())
467	msg.Stats = coord.stats
468	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
469	assert.NotNil(t, coord.pipeline)
470
471	// Slot 1. No bid, so the winner is the boot coordinator
472	// pipelineStarted: true -> false
473	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
474		1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
475	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
476	coord.stats.Sync.Auction.CurrentSlot = slots[1]
477	assert.Equal(t, false, coord.canForge())
478	msg.Stats = coord.stats
479	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
480	assert.Nil(t, coord.pipeline)
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
498func TestCoordinatorStress(t *testing.T) {
499	if os.Getenv("TEST_COORD_STRESS") == "" {
500		return
501	}
502	log.Info("Begin Test Coord Stress")
503	ethClientSetup := test.NewClientSetupExample()
504	ethClientSetup.ChainID = big.NewInt(int64(chainID))
505	var timer timer
506	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
507	modules := newTestModules(t)
508	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
509	syn := newTestSynchronizer(t, ethClient, ethClientSetup, modules)
510
511	coord.Start()
512	ctx, cancel := context.WithCancel(context.Background())
513	var wg sync.WaitGroup
514
515	// Synchronizer loop
516	wg.Add(1)
517	go func() {
518		for {
519			blockData, _, err := syn.Sync2(ctx, nil)
520			if ctx.Err() != nil {
521				wg.Done()
522				return
523			}
524			require.NoError(t, err)
525			if blockData != nil {
526				stats := syn.Stats()
527				coord.SendMsg(ctx, MsgSyncBlock{
528					Stats:   *stats,
529					Batches: blockData.Rollup.Batches,
530					Vars: synchronizer.SCVariablesPtr{
531						Rollup:   blockData.Rollup.Vars,
532						Auction:  blockData.Auction.Vars,
533						WDelayer: blockData.WDelayer.Vars,
534					},
535				})
536			} else {
537				time.Sleep(100 * time.Millisecond)
538			}
539		}
540	}()
541
542	// Blockchain mining loop
543	wg.Add(1)
544	go func() {
545		for {
546			select {
547			case <-ctx.Done():
548				wg.Done()
549				return
550			case <-time.After(1 * time.Second):
551				ethClient.CtlMineBlock()
552			}
553		}
554	}()
555
556	time.Sleep(600 * time.Second)
557
558	cancel()
559	wg.Wait()
560	coord.Stop()
561}
562
563// TODO: Test Reorg
564// TODO: Test TxMonitor
565// TODO: Test forgeBatch
566// TODO: Test waitServerProof
567// TODO: Test handleReorg