Merge pull request #498 from hermeznetwork/feature/update-txman

Update coordinator to work better under real net

coordinator/coordinator.go

@@ -81,7 +81,7 @@ type Coordinator struct {
 	provers          []prover.Client
 	consts           synchronizer.SCConsts
 	vars             synchronizer.SCVariables
-	stats            *synchronizer.Stats
+	stats            synchronizer.Stats
 	started          bool
 
 	cfg Config
@@ -153,15 +153,22 @@ func NewCoordinator(cfg Config,
 
 		purger: &purger,
 
-		// ethClient: ethClient,
-
 		msgCh: make(chan interface{}),
 		ctx:   ctx,
 		// wg
 		cancel: cancel,
 	}
-	txManager := NewTxManager(&cfg, ethClient, l2DB, &c)
+	ctxTimeout, ctxTimeoutCancel := context.WithTimeout(ctx, 1*time.Second)
+	defer ctxTimeoutCancel()
+	txManager, err := NewTxManager(ctxTimeout, &cfg, ethClient, l2DB, &c,
+		scConsts, initSCVars)
+	if err != nil {
+		return nil, tracerr.Wrap(err)
+	}
 	c.txManager = txManager
+	// Set Eth LastBlockNum to -1 in stats so that stats.Synced() is
+	// guaranteed to return false before it's updated with a real stats
+	c.stats.Eth.LastBlock.Num = -1
 	return &c, nil
 }
 
@@ -191,8 +198,11 @@ type MsgStopPipeline struct {
 }
 
 // SendMsg is a thread safe method to pass a message to the Coordinator
-func (c *Coordinator) SendMsg(msg interface{}) {
-	c.msgCh <- msg
+func (c *Coordinator) SendMsg(ctx context.Context, msg interface{}) {
+	select {
+	case c.msgCh <- msg:
+	case <-ctx.Done():
+	}
 }
 
 func (c *Coordinator) syncSCVars(vars synchronizer.SCVariablesPtr) {
@@ -207,24 +217,42 @@ func (c *Coordinator) syncSCVars(vars synchronizer.SCVariablesPtr) {
 	}
 }
 
-func (c *Coordinator) canForge(stats *synchronizer.Stats) bool {
+func canForge(auctionConstants *common.AuctionConstants, auctionVars *common.AuctionVariables,
+	currentSlot *common.Slot, nextSlot *common.Slot, addr ethCommon.Address, blockNum int64) bool {
+	var slot *common.Slot
+	if currentSlot.StartBlock <= blockNum && blockNum <= currentSlot.EndBlock {
+		slot = currentSlot
+	} else if nextSlot.StartBlock <= blockNum && blockNum <= nextSlot.EndBlock {
+		slot = nextSlot
+	} else {
+		log.Warnw("Coordinator: requested blockNum for canForge is outside slot",
+			"blockNum", blockNum, "currentSlot", currentSlot,
+			"nextSlot", nextSlot,
+		)
+		return false
+	}
 	anyoneForge := false
-	if !stats.Sync.Auction.CurrentSlot.ForgerCommitment &&
-		c.consts.Auction.RelativeBlock(stats.Eth.LastBlock.Num+1) >= int64(c.vars.Auction.SlotDeadline) {
+	if !slot.ForgerCommitment &&
+		auctionConstants.RelativeBlock(blockNum) >= int64(auctionVars.SlotDeadline) {
 		log.Debugw("Coordinator: anyone can forge in the current slot (slotDeadline passed)",
-			"block", stats.Eth.LastBlock.Num+1)
+			"block", blockNum)
 		anyoneForge = true
 	}
-	if stats.Sync.Auction.CurrentSlot.Forger == c.cfg.ForgerAddress || anyoneForge {
+	if slot.Forger == addr || anyoneForge {
 		return true
 	}
 	return false
 }
 
-func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
-	c.txManager.SetLastBlock(stats.Eth.LastBlock.Num)
+func (c *Coordinator) canForge() bool {
+	blockNum := c.stats.Eth.LastBlock.Num + 1
+	return canForge(&c.consts.Auction, &c.vars.Auction,
+		&c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
+		c.cfg.ForgerAddress, blockNum)
+}
 
-	canForge := c.canForge(stats)
+func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
+	canForge := c.canForge()
 	if c.pipeline == nil {
 		if canForge {
 			log.Infow("Coordinator: forging state begin", "block",
@@ -274,22 +302,24 @@ func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats)
 }
 
 func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
-	c.stats = &msg.Stats
+	c.stats = msg.Stats
 	c.syncSCVars(msg.Vars)
+	c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
 	if c.pipeline != nil {
-		c.pipeline.SetSyncStatsVars(&msg.Stats, &msg.Vars)
+		c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
 	}
 	if !c.stats.Synced() {
 		return nil
 	}
-	return c.syncStats(ctx, c.stats)
+	return c.syncStats(ctx, &c.stats)
 }
 
 func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
-	c.stats = &msg.Stats
+	c.stats = msg.Stats
 	c.syncSCVars(msg.Vars)
+	c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
 	if c.pipeline != nil {
-		c.pipeline.SetSyncStatsVars(&msg.Stats, &msg.Vars)
+		c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
 	}
 	if common.BatchNum(c.stats.Sync.LastBatch) < c.pipelineBatchNum {
 		// There's been a reorg and the batch from which the pipeline
@@ -373,11 +403,11 @@ func (c *Coordinator) Start() {
 				}
 				waitDuration = longWaitDuration
 			case <-time.After(waitDuration):
-				if c.stats == nil {
+				if !c.stats.Synced() {
 					waitDuration = longWaitDuration
 					continue
 				}
-				if err := c.syncStats(c.ctx, c.stats); c.ctx.Err() != nil {
+				if err := c.syncStats(c.ctx, &c.stats); c.ctx.Err() != nil {
 					continue
 				} else if err != nil {
 					log.Errorw("Coordinator.syncStats", "err", err)

coordinator/coordinator_test.go

@@ -233,12 +233,17 @@ func TestCoordinatorFlow(t *testing.T) {
 	// Bid for slot 2 and 4
 	_, err := ethClient.AuctionSetCoordinator(forger, "https://foo.bar")
 	require.NoError(t, err)
-	_, err = ethClient.AuctionBidSimple(2, big.NewInt(9999))
+	bid, ok := new(big.Int).SetString("11000000000000000000", 10)
+	if !ok {
+		panic("bad bid")
+	}
+	_, err = ethClient.AuctionBidSimple(3, bid)
 	require.NoError(t, err)
-	_, err = ethClient.AuctionBidSimple(4, big.NewInt(9999))
+	_, err = ethClient.AuctionBidSimple(5, bid)
 	require.NoError(t, err)
 
 	coord.Start()
+	ctx := context.Background()
 	time.Sleep(1 * time.Second)
 
 	waitForSlot := func(slot int64) {
@@ -259,9 +264,17 @@ func TestCoordinatorFlow(t *testing.T) {
 			stats.Sync.LastBatch = stats.Eth.LastBatch
 			canForge, err := ethClient.AuctionCanForge(forger, blockNum+1)
 			require.NoError(t, err)
+			var slot common.Slot
+			slotNum := ethClientSetup.AuctionConstants.SlotNum(blockNum + 1)
+			slot.StartBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
+				(slotNum)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
+			slot.EndBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
+				(slotNum+1)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
 			if canForge {
-				stats.Sync.Auction.CurrentSlot.Forger = forger
+				slot.Forger = forger
 			}
+			stats.Sync.Auction.CurrentSlot = slot
+
 			// Copy stateDB to synchronizer if there was a new batch
 			source := fmt.Sprintf("%v/BatchNum%v", batchBuilderDBPath, stats.Sync.LastBatch)
 			dest := fmt.Sprintf("%v/BatchNum%v", syncDBPath, stats.Sync.LastBatch)
@@ -273,7 +286,7 @@ func TestCoordinatorFlow(t *testing.T) {
 					require.NoError(t, err)
 				}
 			}
-			coord.SendMsg(MsgSyncBlock{
+			coord.SendMsg(ctx, MsgSyncBlock{
 				Stats: stats,
 			})
 		}
@@ -329,7 +342,7 @@ func TestCoordCanForge(t *testing.T) {
 	if !ok {
 		panic("bad bid")
 	}
-	_, err = ethClient.AuctionBidSimple(2, bid)
+	_, err = ethClient.AuctionBidSimple(3, bid)
 	require.NoError(t, err)
 
 	modules2 := newTestModules(t)
@@ -343,28 +356,48 @@ func TestCoordCanForge(t *testing.T) {
 
 	var stats synchronizer.Stats
 
+	slots := [4]common.Slot{}
+	for i := 0; i < 4; i++ {
+		slots[i].StartBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
+			int64(i)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
+		slots[i].EndBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
+			int64(i+1)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
+		// Only slot 3 has Coordinator winner, the rest are BootCoordinator
+		if i == 3 {
+			slots[i].Forger = forger
+		} else {
+			slots[i].Forger = bootForger
+		}
+	}
+
 	// Slot 0.  No bid, so the winner is the boot coordinator
 	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
 	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = bootForger
-	assert.Equal(t, false, coord.canForge(&stats))
-	assert.Equal(t, true, bootCoord.canForge(&stats))
+	stats.Sync.Auction.CurrentSlot = slots[0]
+	coord.stats = stats
+	bootCoord.stats = stats
+	assert.Equal(t, false, coord.canForge())
+	assert.Equal(t, true, bootCoord.canForge())
 
 	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
 	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
 		int64(ethClientSetup.AuctionVariables.SlotDeadline)
 	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = bootForger
-	assert.Equal(t, true, coord.canForge(&stats))
-	assert.Equal(t, true, bootCoord.canForge(&stats))
+	stats.Sync.Auction.CurrentSlot = slots[0]
+	coord.stats = stats
+	bootCoord.stats = stats
+	assert.Equal(t, true, coord.canForge())
+	assert.Equal(t, true, bootCoord.canForge())
 
-	// Slot 1. coordinator bid, so the winner is the coordinator
+	// Slot 3. coordinator bid, so the winner is the coordinator
 	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
-		1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
+		3*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
 	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = forger
-	assert.Equal(t, true, coord.canForge(&stats))
-	assert.Equal(t, false, bootCoord.canForge(&stats))
+	stats.Sync.Auction.CurrentSlot = slots[3]
+	coord.stats = stats
+	bootCoord.stats = stats
+	assert.Equal(t, true, coord.canForge())
+	assert.Equal(t, false, bootCoord.canForge())
 }
 
 func TestCoordHandleMsgSyncBlock(t *testing.T) {
@@ -382,49 +415,67 @@ func TestCoordHandleMsgSyncBlock(t *testing.T) {
 	if !ok {
 		panic("bad bid")
 	}
-	_, err = ethClient.AuctionBidSimple(2, bid)
+	_, err = ethClient.AuctionBidSimple(3, bid)
 	require.NoError(t, err)
 
+	slots := [4]common.Slot{}
+	for i := 0; i < 4; i++ {
+		slots[i].StartBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
+			int64(i)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
+		slots[i].EndBlock = ethClientSetup.AuctionConstants.GenesisBlockNum +
+			int64(i+1)*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
+		// Only slot 3 has Coordinator winner, the rest are BootCoordinator
+		if i == 3 {
+			slots[i].Forger = forger
+		} else {
+			slots[i].Forger = bootForger
+		}
+	}
+
 	var msg MsgSyncBlock
-	stats := &msg.Stats
+	coord.stats = msg.Stats
 	ctx := context.Background()
 
 	// Slot 0.  No bid, so the winner is the boot coordinator
 	// pipelineStarted: false -> false
-	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
-	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = bootForger
-	assert.Equal(t, false, coord.canForge(stats))
+	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum
+	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
+	coord.stats.Sync.Auction.CurrentSlot = slots[0]
+	assert.Equal(t, false, coord.canForge())
+	msg.Stats = coord.stats
 	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
 	assert.Nil(t, coord.pipeline)
 
 	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
 	// pipelineStarted: false -> true
-	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
+	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
 		int64(ethClientSetup.AuctionVariables.SlotDeadline)
-	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = bootForger
-	assert.Equal(t, true, coord.canForge(stats))
+	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
+	coord.stats.Sync.Auction.CurrentSlot = slots[0]
+	assert.Equal(t, true, coord.canForge())
+	msg.Stats = coord.stats
 	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
 	assert.NotNil(t, coord.pipeline)
 
 	// Slot 0.  No bid, and we reach the deadline, so anyone can forge
 	// pipelineStarted: true -> true
-	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
+	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
 		int64(ethClientSetup.AuctionVariables.SlotDeadline) + 1
-	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = bootForger
-	assert.Equal(t, true, coord.canForge(stats))
+	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
+	coord.stats.Sync.Auction.CurrentSlot = slots[0]
+	assert.Equal(t, true, coord.canForge())
+	msg.Stats = coord.stats
 	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
 	assert.NotNil(t, coord.pipeline)
 
-	// Slot 0. No bid, so the winner is the boot coordinator
+	// Slot 1. No bid, so the winner is the boot coordinator
 	// pipelineStarted: true -> false
-	stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
+	coord.stats.Eth.LastBlock.Num = ethClientSetup.AuctionConstants.GenesisBlockNum +
 		1*int64(ethClientSetup.AuctionConstants.BlocksPerSlot)
-	stats.Sync.LastBlock = stats.Eth.LastBlock
-	stats.Sync.Auction.CurrentSlot.Forger = bootForger
-	assert.Equal(t, false, coord.canForge(stats))
+	coord.stats.Sync.LastBlock = coord.stats.Eth.LastBlock
+	coord.stats.Sync.Auction.CurrentSlot = slots[1]
+	assert.Equal(t, false, coord.canForge())
+	msg.Stats = coord.stats
 	require.NoError(t, coord.handleMsgSyncBlock(ctx, &msg))
 	assert.Nil(t, coord.pipeline)
 }
@@ -473,7 +524,7 @@ func TestCoordinatorStress(t *testing.T) {
 			require.NoError(t, err)
 			if blockData != nil {
 				stats := syn.Stats()
-				coord.SendMsg(MsgSyncBlock{
+				coord.SendMsg(ctx, MsgSyncBlock{
 					Stats:   *stats,
 					Batches: blockData.Rollup.Batches,
 					Vars: synchronizer.SCVariablesPtr{

coordinator/pipeline.go

@@ -71,7 +71,7 @@ func NewPipeline(ctx context.Context,
 		if err := prover.WaitReady(ctx); err != nil {
 			log.Errorw("prover.WaitReady", "err", err)
 		} else {
-			proversPool.Add(prover)
+			proversPool.Add(ctx, prover)
 			proversPoolSize++
 		}
 	}
@@ -94,8 +94,11 @@ func NewPipeline(ctx context.Context,
 }
 
 // SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
-func (p *Pipeline) SetSyncStatsVars(stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
-	p.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}
+func (p *Pipeline) SetSyncStatsVars(ctx context.Context, stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
+	select {
+	case p.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}:
+	case <-ctx.Done():
+	}
 }
 
 // reset pipeline state
@@ -161,7 +164,7 @@ func (p *Pipeline) handleForgeBatch(ctx context.Context, batchNum common.BatchNu
 	} else if err != nil {
 		log.Errorw("sendServerProof", "err", err)
 		batchInfo.ServerProof = nil
-		p.proversPool.Add(serverProof)
+		p.proversPool.Add(ctx, serverProof)
 		return nil, err
 	}
 	return batchInfo, nil
@@ -202,7 +205,10 @@ func (p *Pipeline) Start(batchNum common.BatchNum,
 					continue
 				} else {
 					p.batchNum = batchNum
-					batchChSentServerProof <- batchInfo
+					select {
+					case batchChSentServerProof <- batchInfo:
+					case <-p.ctx.Done():
+					}
 				}
 			}
 		}
@@ -219,16 +225,15 @@ func (p *Pipeline) Start(batchNum common.BatchNum,
 			case batchInfo := <-batchChSentServerProof:
 				err := p.waitServerProof(p.ctx, batchInfo)
 				// We are done with this serverProof, add it back to the pool
-				p.proversPool.Add(batchInfo.ServerProof)
+				p.proversPool.Add(p.ctx, batchInfo.ServerProof)
 				batchInfo.ServerProof = nil
 				if p.ctx.Err() != nil {
 					continue
-				}
-				if err != nil {
+				} else if err != nil {
 					log.Errorw("waitServerProof", "err", err)
 					continue
 				}
-				p.txManager.AddBatch(batchInfo)
+				p.txManager.AddBatch(p.ctx, batchInfo)
 			}
 		}
 	}()
@@ -365,6 +370,7 @@ func (p *Pipeline) forgeBatch(batchNum common.BatchNum) (batchInfo *BatchInfo, e
 	batchInfo.ZKInputs = zkInputs
 	batchInfo.Debug.Status = StatusForged
 	p.cfg.debugBatchStore(batchInfo)
+	log.Infow("Pipeline: batch forged internally", "batch", batchInfo.BatchNum)
 
 	return batchInfo, nil
 }
@@ -380,6 +386,7 @@ func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) er
 	batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
 	batchInfo.Debug.Status = StatusProof
 	p.cfg.debugBatchStore(batchInfo)
+	log.Infow("Pipeline: batch proof calculated", "batch", batchInfo.BatchNum)
 	return nil
 }
 

coordinator/pipeline_test.go

@@ -291,7 +291,9 @@ func TestEthRollupForgeBatch(t *testing.T) {
 	batchInfo.PublicInputs = pubInputs
 
 	batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
-	_, err = client.RollupForgeBatch(batchInfo.ForgeBatchArgs)
+	auth, err := client.NewAuth()
+	require.NoError(t, err)
+	_, err = client.RollupForgeBatch(batchInfo.ForgeBatchArgs, auth)
 	require.NoError(t, err)
 	batchInfo.Proof = proof
 }

coordinator/proverspool.go

@@ -22,8 +22,11 @@ func NewProversPool(maxServerProofs int) *ProversPool {
 }
 
 // Add a prover to the pool
-func (p *ProversPool) Add(serverProof prover.Client) {
-	p.pool <- serverProof
+func (p *ProversPool) Add(ctx context.Context, serverProof prover.Client) {
+	select {
+	case p.pool <- serverProof:
+	case <-ctx.Done():
+	}
 }
 
 // Get returns the next available prover

coordinator/txmanager.go

@@ -3,14 +3,18 @@ package coordinator
 import (
 	"context"
 	"fmt"
-	"strings"
+	"math/big"
 	"time"
 
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/hermeznetwork/hermez-node/common"
 	"github.com/hermeznetwork/hermez-node/db/l2db"
 	"github.com/hermeznetwork/hermez-node/eth"
 	"github.com/hermeznetwork/hermez-node/log"
+	"github.com/hermeznetwork/hermez-node/synchronizer"
 	"github.com/hermeznetwork/tracerr"
 )
 
@@ -18,61 +22,150 @@ import (
 // call to forge, waits for transaction confirmation, and keeps checking them
 // until a number of confirmed blocks have passed.
 type TxManager struct {
-	cfg         Config
-	ethClient   eth.ClientInterface
-	l2DB        *l2db.L2DB   // Used only to mark forged txs as forged in the L2DB
-	coord       *Coordinator // Used only to send messages to stop the pipeline
-	batchCh     chan *BatchInfo
-	lastBlockCh chan int64
-	queue       []*BatchInfo
-	lastBlock   int64
-	// lastConfirmedBatch stores the last BatchNum that who's forge call was confirmed
-	lastConfirmedBatch common.BatchNum
+	cfg       Config
+	ethClient eth.ClientInterface
+	l2DB      *l2db.L2DB   // Used only to mark forged txs as forged in the L2DB
+	coord     *Coordinator // Used only to send messages to stop the pipeline
+	batchCh   chan *BatchInfo
+	chainID   *big.Int
+	account   accounts.Account
+	consts    synchronizer.SCConsts
+
+	stats       synchronizer.Stats
+	vars        synchronizer.SCVariables
+	statsVarsCh chan statsVars
+
+	queue []*BatchInfo
+	// lastSuccessBatch stores the last BatchNum that who's forge call was confirmed
+	lastSuccessBatch common.BatchNum
+	lastPendingBatch common.BatchNum
+	lastSuccessNonce uint64
+	lastPendingNonce uint64
 }
 
 // NewTxManager creates a new TxManager
-func NewTxManager(cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
-	coord *Coordinator) *TxManager {
+func NewTxManager(ctx context.Context, cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
+	coord *Coordinator, scConsts *synchronizer.SCConsts, initSCVars *synchronizer.SCVariables) (*TxManager, error) {
+	chainID, err := ethClient.EthChainID()
+	if err != nil {
+		return nil, tracerr.Wrap(err)
+	}
+	address, err := ethClient.EthAddress()
+	if err != nil {
+		return nil, tracerr.Wrap(err)
+	}
+	lastSuccessNonce, err := ethClient.EthNonceAt(ctx, *address, nil)
+	if err != nil {
+		return nil, err
+	}
+	lastPendingNonce, err := ethClient.EthPendingNonceAt(ctx, *address)
+	if err != nil {
+		return nil, err
+	}
+	if lastSuccessNonce != lastPendingNonce {
+		return nil, tracerr.Wrap(fmt.Errorf("lastSuccessNonce (%v) != lastPendingNonce (%v)",
+			lastSuccessNonce, lastPendingNonce))
+	}
+	log.Infow("TxManager started", "nonce", lastSuccessNonce)
 	return &TxManager{
 		cfg:         *cfg,
 		ethClient:   ethClient,
 		l2DB:        l2DB,
 		coord:       coord,
 		batchCh:     make(chan *BatchInfo, queueLen),
-		lastBlockCh: make(chan int64, queueLen),
-		lastBlock:   -1,
-	}
+		statsVarsCh: make(chan statsVars, queueLen),
+		account: accounts.Account{
+			Address: *address,
+		},
+		chainID: chainID,
+		consts:  *scConsts,
+
+		vars: *initSCVars,
+
+		lastSuccessNonce: lastSuccessNonce,
+		lastPendingNonce: lastPendingNonce,
+	}, nil
 }
 
 // AddBatch is a thread safe method to pass a new batch TxManager to be sent to
 // the smart contract via the forge call
-func (t *TxManager) AddBatch(batchInfo *BatchInfo) {
-	t.batchCh <- batchInfo
+func (t *TxManager) AddBatch(ctx context.Context, batchInfo *BatchInfo) {
+	select {
+	case t.batchCh <- batchInfo:
+	case <-ctx.Done():
+	}
 }
 
-// SetLastBlock is a thread safe method to pass the lastBlock to the TxManager
-func (t *TxManager) SetLastBlock(lastBlock int64) {
-	t.lastBlockCh <- lastBlock
+// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
+func (t *TxManager) SetSyncStatsVars(ctx context.Context, stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
+	select {
+	case t.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}:
+	case <-ctx.Done():
+	}
 }
 
-func (t *TxManager) callRollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
+func (t *TxManager) syncSCVars(vars synchronizer.SCVariablesPtr) {
+	if vars.Rollup != nil {
+		t.vars.Rollup = *vars.Rollup
+	}
+	if vars.Auction != nil {
+		t.vars.Auction = *vars.Auction
+	}
+	if vars.WDelayer != nil {
+		t.vars.WDelayer = *vars.WDelayer
+	}
+}
+
+// NewAuth generates a new auth object for an ethereum transaction
+func (t *TxManager) NewAuth(ctx context.Context) (*bind.TransactOpts, error) {
+	gasPrice, err := t.ethClient.EthSuggestGasPrice(ctx)
+	if err != nil {
+		return nil, tracerr.Wrap(err)
+	}
+	inc := new(big.Int).Set(gasPrice)
+	const gasPriceDiv = 100
+	inc.Div(inc, new(big.Int).SetUint64(gasPriceDiv))
+	gasPrice.Add(gasPrice, inc)
+	// log.Debugw("TxManager: transaction metadata", "gasPrice", gasPrice)
+
+	auth, err := bind.NewKeyStoreTransactorWithChainID(t.ethClient.EthKeyStore(), t.account, t.chainID)
+	if err != nil {
+		return nil, tracerr.Wrap(err)
+	}
+	auth.Value = big.NewInt(0) // in wei
+	// TODO: Calculate GasLimit based on the contents of the ForgeBatchArgs
+	auth.GasLimit = 1000000
+	auth.GasPrice = gasPrice
+	auth.Nonce = nil
+
+	return auth, nil
+}
+
+func (t *TxManager) sendRollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
+	// TODO: Check if we can forge in the next blockNum, abort if we can't
 	batchInfo.Debug.Status = StatusSent
-	batchInfo.Debug.SendBlockNum = t.lastBlock + 1
+	batchInfo.Debug.SendBlockNum = t.stats.Eth.LastBlock.Num + 1
 	batchInfo.Debug.SendTimestamp = time.Now()
 	batchInfo.Debug.StartToSendDelay = batchInfo.Debug.SendTimestamp.Sub(
 		batchInfo.Debug.StartTimestamp).Seconds()
 	var ethTx *types.Transaction
 	var err error
+	auth, err := t.NewAuth(ctx)
+	if err != nil {
+		return tracerr.Wrap(err)
+	}
+	auth.Nonce = big.NewInt(int64(t.lastPendingNonce))
+	t.lastPendingNonce++
 	for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
-		ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
+		ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs, auth)
 		if err != nil {
-			if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
-				log.Debugw("TxManager ethClient.RollupForgeBatch", "err", err,
-					"block", t.lastBlock+1)
-				return tracerr.Wrap(err)
-			}
+			// if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
+			// 	log.Errorw("TxManager ethClient.RollupForgeBatch", "err", err,
+			// 		"block", t.stats.Eth.LastBlock.Num+1)
+			// 	return tracerr.Wrap(err)
+			// }
 			log.Errorw("TxManager ethClient.RollupForgeBatch",
-				"attempt", attempt, "err", err, "block", t.lastBlock+1,
+				"attempt", attempt, "err", err, "block", t.stats.Eth.LastBlock.Num+1,
 				"batchNum", batchInfo.BatchNum)
 		} else {
 			break
@@ -89,12 +182,15 @@ func (t *TxManager) callRollupForgeBatch(ctx context.Context, batchInfo *BatchIn
 	batchInfo.EthTx = ethTx
 	log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash().Hex())
 	t.cfg.debugBatchStore(batchInfo)
+	t.lastPendingBatch = batchInfo.BatchNum
 	if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
 		return tracerr.Wrap(err)
 	}
 	return nil
 }
 
+// checkEthTransactionReceipt takes the txHash from the BatchInfo and stores
+// the corresponding receipt if found
 func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
 	txHash := batchInfo.EthTx.Hash()
 	var receipt *types.Receipt
@@ -103,8 +199,10 @@ func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *B
 		receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
 		if ctx.Err() != nil {
 			continue
-		}
-		if err != nil {
+		} else if tracerr.Unwrap(err) == ethereum.NotFound {
+			err = nil
+			break
+		} else if err != nil {
 			log.Errorw("TxManager ethClient.EthTransactionReceipt",
 				"attempt", attempt, "err", err)
 		} else {
@@ -124,24 +222,28 @@ func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *B
 	return nil
 }
 
-func (t *TxManager) handleReceipt(batchInfo *BatchInfo) (*int64, error) {
+func (t *TxManager) handleReceipt(ctx context.Context, batchInfo *BatchInfo) (*int64, error) {
 	receipt := batchInfo.Receipt
 	if receipt != nil {
 		if receipt.Status == types.ReceiptStatusFailed {
 			batchInfo.Debug.Status = StatusFailed
 			t.cfg.debugBatchStore(batchInfo)
-			log.Errorw("TxManager receipt status is failed", "receipt", receipt)
-			return nil, tracerr.Wrap(fmt.Errorf("ethereum transaction receipt statis is failed"))
+			_, err := t.ethClient.EthCall(ctx, batchInfo.EthTx, receipt.BlockNumber)
+			log.Warnw("TxManager receipt status is failed", "tx", receipt.TxHash.Hex(),
+				"batch", batchInfo.BatchNum, "block", receipt.BlockNumber.Int64(),
+				"err", err)
+			return nil, tracerr.Wrap(fmt.Errorf(
+				"ethereum transaction receipt status is failed: %w", err))
 		} else if receipt.Status == types.ReceiptStatusSuccessful {
 			batchInfo.Debug.Status = StatusMined
 			batchInfo.Debug.MineBlockNum = receipt.BlockNumber.Int64()
 			batchInfo.Debug.StartToMineBlocksDelay = batchInfo.Debug.MineBlockNum -
 				batchInfo.Debug.StartBlockNum
 			t.cfg.debugBatchStore(batchInfo)
-			if batchInfo.BatchNum > t.lastConfirmedBatch {
-				t.lastConfirmedBatch = batchInfo.BatchNum
+			if batchInfo.BatchNum > t.lastSuccessBatch {
+				t.lastSuccessBatch = batchInfo.BatchNum
 			}
-			confirm := t.lastBlock - receipt.BlockNumber.Int64()
+			confirm := t.stats.Eth.LastBlock.Num - receipt.BlockNumber.Int64()
 			return &confirm, nil
 		}
 	}
@@ -153,25 +255,41 @@ func (t *TxManager) Run(ctx context.Context) {
 	next := 0
 	waitDuration := longWaitDuration
 
+	var statsVars statsVars
+	select {
+	case statsVars = <-t.statsVarsCh:
+	case <-ctx.Done():
+	}
+	t.stats = statsVars.Stats
+	t.syncSCVars(statsVars.Vars)
+	log.Infow("TxManager: received initial statsVars",
+		"block", t.stats.Eth.LastBlock.Num, "batch", t.stats.Eth.LastBatch)
+
 	for {
 		select {
 		case <-ctx.Done():
 			log.Info("TxManager done")
 			return
-		case lastBlock := <-t.lastBlockCh:
-			t.lastBlock = lastBlock
+		case statsVars := <-t.statsVarsCh:
+			t.stats = statsVars.Stats
+			t.syncSCVars(statsVars.Vars)
 		case batchInfo := <-t.batchCh:
-			if err := t.callRollupForgeBatch(ctx, batchInfo); ctx.Err() != nil {
+			if err := t.sendRollupForgeBatch(ctx, batchInfo); ctx.Err() != nil {
 				continue
 			} else if err != nil {
-				t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch call: %v", err)})
+				// If we reach here it's because our ethNode has
+				// been unable to send the transaction to
+				// ethereum.  This could be due to the ethNode
+				// failure, or an invalid transaction (that
+				// can't be mined)
+				t.coord.SendMsg(ctx, MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch send: %v", err)})
 				continue
 			}
-			log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
 			t.queue = append(t.queue, batchInfo)
 			waitDuration = t.cfg.TxManagerCheckInterval
 		case <-time.After(waitDuration):
 			if len(t.queue) == 0 {
+				waitDuration = longWaitDuration
 				continue
 			}
 			current := next
@@ -180,23 +298,33 @@ func (t *TxManager) Run(ctx context.Context) {
 			if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
 				continue
 			} else if err != nil { //nolint:staticcheck
-				// We can't get the receipt for the
-				// transaction, so we can't confirm if it was
-				// mined
-				t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
+				// Our ethNode is giving an error different
+				// than "not found" when getting the receipt
+				// for the transaction, so we can't figure out
+				// if it was not mined, mined and succesfull or
+				// mined and failed.  This could be due to the
+				// ethNode failure.
+				t.coord.SendMsg(ctx, MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
 			}
 
-			confirm, err := t.handleReceipt(batchInfo)
-			if err != nil { //nolint:staticcheck
+			confirm, err := t.handleReceipt(ctx, batchInfo)
+			if ctx.Err() != nil {
+				continue
+			} else if err != nil { //nolint:staticcheck
 				// Transaction was rejected
-				t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
+				t.queue = append(t.queue[:current], t.queue[current+1:]...)
+				if len(t.queue) == 0 {
+					next = 0
+				} else {
+					next = current % len(t.queue)
+				}
+				t.coord.SendMsg(ctx, MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
 			}
 			if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
 				log.Debugw("TxManager tx for RollupForgeBatch confirmed",
 					"batch", batchInfo.BatchNum)
 				t.queue = append(t.queue[:current], t.queue[current+1:]...)
 				if len(t.queue) == 0 {
-					waitDuration = longWaitDuration
 					next = 0
 				} else {
 					next = current % len(t.queue)
@@ -205,3 +333,11 @@ func (t *TxManager) Run(ctx context.Context) {
 		}
 	}
 }
+
+// nolint reason: this function will be used in the future
+//nolint:unused
+func (t *TxManager) canForge(stats *synchronizer.Stats, blockNum int64) bool {
+	return canForge(&t.consts.Auction, &t.vars.Auction,
+		&stats.Sync.Auction.CurrentSlot, &stats.Sync.Auction.NextSlot,
+		t.cfg.ForgerAddress, blockNum)
+}