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