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	_, err = ethClient.AuctionBidSimple(2, big.NewInt(9999))
237	require.NoError(t, err)
238	_, err = ethClient.AuctionBidSimple(4, big.NewInt(9999))
239	require.NoError(t, err)
240
241	coord.Start()
242	time.Sleep(1 * time.Second)
243
244	waitForSlot := func(slot int64) {
245		for {
246			blockNum, err := ethClient.EthLastBlock()
247			require.NoError(t, err)
248			nextBlockSlot, err := ethClient.AuctionGetSlotNumber(blockNum + 1)
249			require.NoError(t, err)
250			if nextBlockSlot == slot {
251				break
252			}
253			ethClient.CtlMineBlock()
254			time.Sleep(100 * time.Millisecond)
255			var stats synchronizer.Stats
256			stats.Eth.LastBlock = *ethClient.CtlLastBlock()
257			stats.Sync.LastBlock = stats.Eth.LastBlock
258			stats.Eth.LastBatch = ethClient.CtlLastForgedBatch()
259			stats.Sync.LastBatch = stats.Eth.LastBatch
260			canForge, err := ethClient.AuctionCanForge(forger, blockNum+1)
261			require.NoError(t, err)
262			if canForge {
263				stats.Sync.Auction.CurrentSlot.Forger = forger
264			}
265			// Copy stateDB to synchronizer if there was a new batch
266			source := fmt.Sprintf("%v/BatchNum%v", batchBuilderDBPath, stats.Sync.LastBatch)
267			dest := fmt.Sprintf("%v/BatchNum%v", syncDBPath, stats.Sync.LastBatch)
268			if stats.Sync.LastBatch != 0 {
269				if _, err := os.Stat(dest); os.IsNotExist(err) {
270					log.Infow("Making pebble checkpoint for sync",
271						"source", source, "dest", dest)
272					err = pebbleMakeCheckpoint(source, dest)
273					require.NoError(t, err)
274				}
275			}
276			coord.SendMsg(MsgSyncBlock{
277				Stats: stats,
278			})
279		}
280	}
281
282	// NOTE: With the current test, the coordinator will enter in forge
283	// time before the bidded slot because no one else is forging in the
284	// other slots before the slot deadline.
285	// simulate forgeSequence time
286	waitForSlot(2)
287	log.Info("~~~ simulate entering in forge time")
288	time.Sleep(1 * time.Second)
289
290	// simulate going out from forgeSequence
291	waitForSlot(3)
292	log.Info("~~~ simulate going out from forge time")
293	time.Sleep(1 * time.Second)
294
295	// simulate entering forgeSequence time again
296	waitForSlot(4)
297	log.Info("~~~ simulate entering in forge time again")
298	time.Sleep(2 * time.Second)
299
300	// simulate stopping forgerLoop by channel
301	log.Info("~~~ simulate stopping forgerLoop by closing coordinator stopch")
302	coord.Stop()
303	time.Sleep(1 * time.Second)
304}
305
306func TestCoordinatorStartStop(t *testing.T) {
307	ethClientSetup := test.NewClientSetupExample()
308	ethClientSetup.ChainID = big.NewInt(int64(chainID))
309	var timer timer
310	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
311	modules := newTestModules(t)
312	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
313	coord.Start()
314	coord.Stop()
315}
316
317func TestCoordCanForge(t *testing.T) {
318	ethClientSetup := test.NewClientSetupExample()
319	ethClientSetup.ChainID = big.NewInt(int64(chainID))
320	bootForger := ethClientSetup.AuctionVariables.BootCoordinator
321
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	_, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
327	require.NoError(t, err)
328	bid, ok := new(big.Int).SetString("12000000000000000000", 10)
329	if !ok {
330		panic("bad bid")
331	}
332	_, err = ethClient.AuctionBidSimple(2, bid)
333	require.NoError(t, err)
334
335	modules2 := newTestModules(t)
336	bootCoord := newTestCoordinator(t, bootForger, ethClient, ethClientSetup, modules2)
337
338	assert.Equal(t, forger, coord.cfg.ForgerAddress)
339	assert.Equal(t, bootForger, bootCoord.cfg.ForgerAddress)
340	ethBootCoord, err := ethClient.AuctionGetBootCoordinator()
341	require.NoError(t, err)
342	assert.Equal(t, &bootForger, ethBootCoord)
343
344	var stats synchronizer.Stats
345
346	// Slot 0.  No bid, so the winner is the boot coordinator
347	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
348	stats.Sync.LastBlock = stats.Eth.LastBlock
349	stats.Sync.Auction.CurrentSlot.Forger = bootForger
350	assert.Equal(t, false, coord.canForge(&stats))
351	assert.Equal(t, true, bootCoord.canForge(&stats))
352
353	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
354	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
355		int64(ethClientSetup.AuctionVariables.SlotDeadline)
356	stats.Sync.LastBlock = stats.Eth.LastBlock
357	stats.Sync.Auction.CurrentSlot.Forger = bootForger
358	assert.Equal(t, true, coord.canForge(&stats))
359	assert.Equal(t, true, bootCoord.canForge(&stats))
360
361	// Slot 1. coordinator bid, so the winner is the coordinator
362	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
363		1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
364	stats.Sync.LastBlock = stats.Eth.LastBlock
365	stats.Sync.Auction.CurrentSlot.Forger = forger
366	assert.Equal(t, true, coord.canForge(&stats))
367	assert.Equal(t, false, bootCoord.canForge(&stats))
368}
369
370func TestCoordHandleMsgSyncBlock(t *testing.T) {
371	ethClientSetup := test.NewClientSetupExample()
372	ethClientSetup.ChainID = big.NewInt(int64(chainID))
373	bootForger := ethClientSetup.AuctionVariables.BootCoordinator
374
375	var timer timer
376	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
377	modules := newTestModules(t)
378	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
379	_, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
380	require.NoError(t, err)
381	bid, ok := new(big.Int).SetString("11000000000000000000", 10)
382	if !ok {
383		panic("bad bid")
384	}
385	_, err = ethClient.AuctionBidSimple(2, bid)
386	require.NoError(t, err)
387
388	var msg MsgSyncBlock
389	stats := &msg.Stats
390	ctx := context.Background()
391
392	// Slot 0.  No bid, so the winner is the boot coordinator
393	// pipelineStarted: false -> false
394	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
395	stats.Sync.LastBlock = stats.Eth.LastBlock
396	stats.Sync.Auction.CurrentSlot.Forger = bootForger
397	assert.Equal(t, false, coord.canForge(stats))
398	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
399	assert.Nil(t, coord.pipeline)
400
401	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
402	// pipelineStarted: false -> true
403	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
404		int64(ethClientSetup.AuctionVariables.SlotDeadline)
405	stats.Sync.LastBlock = stats.Eth.LastBlock
406	stats.Sync.Auction.CurrentSlot.Forger = bootForger
407	assert.Equal(t, true, coord.canForge(stats))
408	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
409	assert.NotNil(t, coord.pipeline)
410
411	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
412	// pipelineStarted: true -> true
413	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
414		int64(ethClientSetup.AuctionVariables.SlotDeadline) + 1
415	stats.Sync.LastBlock = stats.Eth.LastBlock
416	stats.Sync.Auction.CurrentSlot.Forger = bootForger
417	assert.Equal(t, true, coord.canForge(stats))
418	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
419	assert.NotNil(t, coord.pipeline)
420
421	// Slot 0. No bid, so the winner is the boot coordinator
422	// pipelineStarted: true -> false
423	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
424		1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
425	stats.Sync.LastBlock = stats.Eth.LastBlock
426	stats.Sync.Auction.CurrentSlot.Forger = bootForger
427	assert.Equal(t, false, coord.canForge(stats))
428	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
429	assert.Nil(t, coord.pipeline)
430}
431
432// ethAddTokens adds the tokens from the blocks to the blockchain
433func ethAddTokens(blocks []common.BlockData, client *test.Client) {
434	for _, block := range blocks {
435		for _, token := range block.Rollup.AddedTokens {
436			consts := eth.ERC20Consts{
437				Name:     fmt.Sprintf("Token %d", token.TokenID),
438				Symbol:   fmt.Sprintf("TK%d", token.TokenID),
439				Decimals: 18,
440			}
441			// tokenConsts[token.TokenID] = consts
442			client.CtlAddERC20(token.EthAddr, consts)
443		}
444	}
445}
446
447func TestCoordinatorStress(t *testing.T) {
448	if os.Getenv("TEST_COORD_STRESS") == "" {
449		return
450	}
451	log.Info("Begin Test Coord Stress")
452	ethClientSetup := test.NewClientSetupExample()
453	ethClientSetup.ChainID = big.NewInt(int64(chainID))
454	var timer timer
455	ethClient := test.NewClient(true, &timer, &bidder, ethClientSetup)
456	modules := newTestModules(t)
457	coord := newTestCoordinator(t, forger, ethClient, ethClientSetup, modules)
458	syn := newTestSynchronizer(t, ethClient, ethClientSetup, modules)
459
460	coord.Start()
461	ctx, cancel := context.WithCancel(context.Background())
462	var wg sync.WaitGroup
463
464	// Synchronizer loop
465	wg.Add(1)
466	go func() {
467		for {
468			blockData, _, err := syn.Sync2(ctx, nil)
469			if ctx.Err() != nil {
470				wg.Done()
471				return
472			}
473			require.NoError(t, err)
474			if blockData != nil {
475				stats := syn.Stats()
476				coord.SendMsg(MsgSyncBlock{
477					Stats:   *stats,
478					Batches: blockData.Rollup.Batches,
479					Vars: synchronizer.SCVariablesPtr{
480						Rollup:   blockData.Rollup.Vars,
481						Auction:  blockData.Auction.Vars,
482						WDelayer: blockData.WDelayer.Vars,
483					},
484				})
485			} else {
486				time.Sleep(100 * time.Millisecond)
487			}
488		}
489	}()
490
491	// Blockchain mining loop
492	wg.Add(1)
493	go func() {
494		for {
495			select {
496			case <-ctx.Done():
497				wg.Done()
498				return
499			case <-time.After(1 * time.Second):
500				ethClient.CtlMineBlock()
501			}
502		}
503	}()
504
505	time.Sleep(600 * time.Second)
506
507	cancel()
508	wg.Wait()
509	coord.Stop()
510}
511
512// TODO: Test Reorg
513// TODO: Test TxMonitor
514// TODO: Test forgeBatch
515// TODO: Test waitServerProof
516// TODO: Test handleReorg