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