Merge branch 'main' into adlerjohn/readme-build-from-source

.github/workflows/build-evmos.yml

@@ -3,12 +3,10 @@ name: build-evmos
 on:
   push:
     branches:
-      - '**'
-    tags:
-      - 'v*.*.*'
+      - main
   pull_request:
-    branches:
-      - 'main'
+  release:
+    types: [published]
   workflow_dispatch:
 
 env:
@@ -26,7 +24,7 @@ jobs:
       - name: "Checkout source code"
         uses: "actions/checkout@v3"
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v3
         with:
           go-version: 1.17
       - name: up a level

.github/workflows/lint.yml

@@ -1,18 +1,19 @@
-name: golangci-lint
+name: Linters
 on:
   push:
     tags:
       - v*
     branches:
       - main
   pull_request:
+
 jobs:
   golangci:
     name: lint
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
-      - uses: actions/setup-go@v2
+      - uses: actions/setup-go@v3
         with:
           go-version: 1.17
       - name: golangci-lint
@@ -38,3 +39,12 @@ jobs:
 
           # Optional: if set to true then the action don't cache or restore ~/.cache/go-build.
           # skip-build-cache: true
+  markdownlint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: markdownlint-cli
+        uses: nosborn/[email protected]
+        with:
+          files: .
+          config-file: .markdownlint.yaml

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
     - uses: actions/checkout@v3
 
     - name: Set up Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
 

.markdownlint.yaml

@@ -0,0 +1,7 @@
+default: true
+MD010:
+  code_blocks: false
+MD013: false
+MD024:
+  allow_different_nesting: true
+

CHANGELOG.md

@@ -5,6 +5,7 @@
 Minor bugfix release, to ensure that Optimint uses the same version of gRPC as Cosmos SDK.
 
 ### BUG FIXES
+
 - Use google.golang.org/grpc v1.33.2 to be compatible with cosmos-sdk ([#315](https://github.com/celestiaorg/optimint/pull/315)) [@jbowen93](https://github.com/jbowen93/)
 - Make TestValidatorSetHandling even more stable ([#314](https://github.com/celestiaorg/optimint/pull/314)) [@tzdybal](https://github.com/tzdybal/)
 
@@ -14,6 +15,7 @@ This is the first Optimint release.
 Optimint supports all ABCI methods and all Tendermint RPCs.
 
 ### FEATURES
+
 - Minimal implementation of ConsensusParams method  ([#292](https://github.com/celestiaorg/optimint/pull/292)) [@tzdybal](https://github.com/tzdybal/)
 - Implement GenesisChunked method  ([#287](https://github.com/celestiaorg/optimint/pull/287)) [@mauriceLC92](https://github.com/mauriceLC92/)
 - Minimalistic validator set handling  ([#286](https://github.com/celestiaorg/optimint/pull/286)) [@tzdybal](https://github.com/tzdybal/)
@@ -61,6 +63,7 @@ Optimint supports all ABCI methods and all Tendermint RPCs.
 - Add design doc to readme  ([#9](https://github.com/celestiaorg/optimint/pull/9)) [@musalbas](https://github.com/musalbas/)
 
 ### IMPROVEMENTS
+
 - Remove extra variable  ([#280](https://github.com/celestiaorg/optimint/pull/280)) [@Raneet10](https://github.com/Raneet10/)
 - Replace tm-db dependency with store package  ([#268](https://github.com/celestiaorg/optimint/pull/268)) [@tzdybal](https://github.com/tzdybal/)
 - Use enum instead of strings for DB type  ([#259](https://github.com/celestiaorg/optimint/pull/259)) [@adlerjohn](https://github.com/adlerjohn/)
@@ -87,6 +90,7 @@ Optimint supports all ABCI methods and all Tendermint RPCs.
 - Use addresses in multiaddr format.  ([#19](https://github.com/celestiaorg/optimint/pull/19)) [@tzdybal](https://github.com/tzdybal/)
 
 ### BUG FIXES
+
 - fix: make `TestValidatorSetHandling` stable ([#313](https://github.com/celestiaorg/optimint/pull/313)) [@tzdybal](https://github.com/tzdybal/)
 - Fix linter on `main`  ([#308](https://github.com/celestiaorg/optimint/pull/308)) [@tzdybal](https://github.com/tzdybal/)
 - Fix multiple bugs for Ethermint  ([#305](https://github.com/celestiaorg/optimint/pull/305)) [@tzdybal](https://github.com/tzdybal/)
@@ -101,4 +105,4 @@ Optimint supports all ABCI methods and all Tendermint RPCs.
 - Fix typos in node/node.go  ([#86](https://github.com/celestiaorg/optimint/pull/86)) [@tzdybal](https://github.com/tzdybal/)
 - Fixing linter errors  ([#55](https://github.com/celestiaorg/optimint/pull/55)) [@tzdybal](https://github.com/tzdybal/)
 - Add peer discovery  ([#17](https://github.com/celestiaorg/optimint/pull/17)) [@tzdybal](https://github.com/tzdybal/)
-- P2P bootstrapping  ([#14](https://github.com/celestiaorg/optimint/pull/14)) [@tzdybal](https://github.com/tzdybal/)
+- P2P bootstrapping  ([#14](https://github.com/celestiaorg/optimint/pull/14)) [@tzdybal](https://github.com/tzdybal/)

README.md

@@ -2,7 +2,7 @@
 
 ABCI-client implementation for Optimistic Rollups.
 
-Design document: https://docs.google.com/document/d/12gZow_JTJjRrmaD2mNTmYniLhyxVLSyDd7Fbxo5UnA8/edit?usp=sharing
+Design document: <https://docs.google.com/document/d/12gZow_JTJjRrmaD2mNTmYniLhyxVLSyDd7Fbxo5UnA8/edit?usp=sharing>
 
 [![build-and-test](https://github.com/celestiaorg/optimint/actions/workflows/test.yml/badge.svg)](https://github.com/celestiaorg/optimint/actions/workflows/test.yml)
 [![golangci-lint](https://github.com/celestiaorg/optimint/actions/workflows/lint.yml/badge.svg)](https://github.com/celestiaorg/optimint/actions/workflows/lint.yml)

block/manager.go

@@ -3,6 +3,7 @@ package block
 import (
 	"context"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -12,6 +13,7 @@ import (
 	"github.com/tendermint/tendermint/crypto/merkle"
 	"github.com/tendermint/tendermint/proxy"
 	tmtypes "github.com/tendermint/tendermint/types"
+	"go.uber.org/multierr"
 
 	"github.com/celestiaorg/optimint/config"
 	"github.com/celestiaorg/optimint/da"
@@ -22,6 +24,9 @@ import (
 	"github.com/celestiaorg/optimint/types"
 )
 
+// defaultDABlockTime is used only if DABlockTime is not configured for manager
+const defaultDABlockTime = 30 * time.Second
+
 // Manager is responsible for aggregating transactions into blocks.
 type Manager struct {
 	lastState state.State
@@ -36,15 +41,21 @@ type Manager struct {
 
 	dalc      da.DataAvailabilityLayerClient
 	retriever da.BlockRetriever
+	// daHeight is the height of the latest processed DA block
+	daHeight uint64
 
 	HeaderOutCh chan *types.Header
 	HeaderInCh  chan *types.Header
 
 	syncTarget uint64
 	blockInCh  chan *types.Block
-	retrieveCh chan uint64
 	syncCache  map[uint64]*types.Block
 
+	// retrieveMtx is used by retrieveCond
+	retrieveMtx *sync.Mutex
+	// retrieveCond is used to notify sync goroutine (SyncLoop) that it needs to retrieve data
+	retrieveCond *sync.Cond
+
 	logger log.Logger
 }
 
@@ -72,12 +83,20 @@ func NewManager(
 	if err != nil {
 		return nil, err
 	}
+	if s.DAHeight < conf.DAStartHeight {
+		s.DAHeight = conf.DAStartHeight
+	}
 
 	proposerAddress, err := getAddress(proposerKey)
 	if err != nil {
 		return nil, err
 	}
 
+	if conf.DABlockTime == 0 {
+		logger.Info("WARNING: using default DA block time", "DABlockTime", defaultDABlockTime)
+		conf.DABlockTime = defaultDABlockTime
+	}
+
 	exec := state.NewBlockExecutor(proposerAddress, conf.NamespaceID, genesis.ChainID, mempool, proxyApp, eventBus, logger)
 	if s.LastBlockHeight+1 == genesis.InitialHeight {
 		res, err := exec.InitChain(genesis)
@@ -100,13 +119,16 @@ func NewManager(
 		executor:    exec,
 		dalc:        dalc,
 		retriever:   dalc.(da.BlockRetriever), // TODO(tzdybal): do it in more gentle way (after MVP)
-		HeaderOutCh: make(chan *types.Header),
-		HeaderInCh:  make(chan *types.Header),
-		blockInCh:   make(chan *types.Block),
-		retrieveCh:  make(chan uint64),
+		daHeight:    s.DAHeight,
+		// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
+		HeaderOutCh: make(chan *types.Header, 100),
+		HeaderInCh:  make(chan *types.Header, 100),
+		blockInCh:   make(chan *types.Block, 100),
+		retrieveMtx: new(sync.Mutex),
 		syncCache:   make(map[uint64]*types.Block),
 		logger:      logger,
 	}
+	agg.retrieveCond = sync.NewCond(agg.retrieveMtx)
 
 	return agg, nil
 }
@@ -142,8 +164,11 @@ func (m *Manager) AggregationLoop(ctx context.Context) {
 }
 
 func (m *Manager) SyncLoop(ctx context.Context) {
+	daTicker := time.NewTicker(m.conf.DABlockTime)
 	for {
 		select {
+		case <-daTicker.C:
+			m.retrieveCond.Signal()
 		case header := <-m.HeaderInCh:
 			m.logger.Debug("block header received", "height", header.Height, "hash", header.Hash())
 			newHeight := header.Height
@@ -153,14 +178,15 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 			// it's handled gently in RetrieveLoop
 			if newHeight > currentHeight {
 				atomic.StoreUint64(&m.syncTarget, newHeight)
-				m.retrieveCh <- newHeight
+				m.retrieveCond.Signal()
 			}
 		case block := <-m.blockInCh:
 			m.logger.Debug("block body retrieved from DALC",
 				"height", block.Header.Height,
 				"hash", block.Hash(),
 			)
 			m.syncCache[block.Header.Height] = block
+			m.retrieveCond.Signal()
 			currentHeight := m.store.Height() // TODO(tzdybal): maybe store a copy in memory
 			b1, ok1 := m.syncCache[currentHeight+1]
 			b2, ok2 := m.syncCache[currentHeight+2]
@@ -181,6 +207,7 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 					continue
 				}
 
+				newState.DAHeight = atomic.LoadUint64(&m.daHeight)
 				m.lastState = newState
 				err = m.store.UpdateState(m.lastState)
 				if err != nil {
@@ -195,50 +222,75 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 	}
 }
 
+// RetrieveLoop is responsible for interacting with DA layer.
 func (m *Manager) RetrieveLoop(ctx context.Context) {
+	// waitCh is used to signal the retrieve loop, that it should process next blocks
+	// retrieveCond can be signalled in completely async manner, and goroutine below
+	// works as some kind of "buffer" for those signals
+	waitCh := make(chan interface{})
+	go func() {
+		for {
+			m.retrieveMtx.Lock()
+			m.retrieveCond.Wait()
+			waitCh <- nil
+			m.retrieveMtx.Unlock()
+			if ctx.Err() != nil {
+				return
+			}
+		}
+	}()
+
 	for {
 		select {
-		case <-m.retrieveCh:
-			target := atomic.LoadUint64(&m.syncTarget)
-			for h := m.store.Height() + 1; h <= target; h++ {
-				m.logger.Debug("trying to retrieve block from DALC", "height", h)
-				m.mustRetrieveBlock(ctx, h)
+		case <-waitCh:
+			for {
+				daHeight := atomic.LoadUint64(&m.daHeight)
+				m.logger.Debug("retrieve", "daHeight", daHeight)
+				err := m.processNextDABlock()
+				if err != nil {
+					m.logger.Error("failed to retrieve block from DALC", "daHeight", daHeight, "errors", err.Error())
+					break
+				}
+				atomic.AddUint64(&m.daHeight, 1)
 			}
 		case <-ctx.Done():
 			return
 		}
 	}
 }
 
-func (m *Manager) mustRetrieveBlock(ctx context.Context, height uint64) {
+func (m *Manager) processNextDABlock() error {
 	// TODO(tzdybal): extract configuration option
 	maxRetries := 10
+	daHeight := atomic.LoadUint64(&m.daHeight)
 
+	var err error
+	m.logger.Debug("trying to retrieve block from DA", "daHeight", daHeight)
 	for r := 0; r < maxRetries; r++ {
-		err := m.fetchBlock(ctx, height)
-		if err == nil {
-			return
+		blockResp, fetchErr := m.fetchBlock(daHeight)
+		if fetchErr != nil {
+			err = multierr.Append(err, fetchErr)
+			time.Sleep(100 * time.Millisecond)
+		} else {
+			for _, block := range blockResp.Blocks {
+				m.blockInCh <- block
+			}
+			return nil
 		}
-		// TODO(tzdybal): configuration option
-		// TODO(tzdybal): exponential backoff
-		time.Sleep(100 * time.Millisecond)
 	}
-	// TODO(tzdybal): this is only temporary solution, for MVP
-	panic("failed to retrieve block with DALC")
+	return err
 }
 
-func (m *Manager) fetchBlock(ctx context.Context, height uint64) error {
+func (m *Manager) fetchBlock(daHeight uint64) (da.ResultRetrieveBlocks, error) {
 	var err error
-	blockRes := m.retriever.RetrieveBlock(height)
+	blockRes := m.retriever.RetrieveBlocks(daHeight)
 	switch blockRes.Code {
-	case da.StatusSuccess:
-		m.blockInCh <- blockRes.Block
 	case da.StatusError:
 		err = fmt.Errorf("failed to retrieve block: %s", blockRes.Message)
 	case da.StatusTimeout:
 		err = fmt.Errorf("timeout during retrieve block: %s", blockRes.Message)
 	}
-	return err
+	return blockRes, err
 }
 
 func (m *Manager) getRemainingSleep(start time.Time) time.Duration {
@@ -305,6 +357,7 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 		return err
 	}
 
+	newState.DAHeight = atomic.LoadUint64(&m.daHeight)
 	m.lastState = newState
 	err = m.store.UpdateState(m.lastState)
 	if err != nil {
@@ -320,6 +373,7 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 }
 
 func (m *Manager) broadcastBlock(ctx context.Context, block *types.Block) error {
+	m.logger.Debug("submitting block to DA layer", "height", block.Header.Height)
 	res := m.dalc.SubmitBlock(block)
 	if res.Code != da.StatusSuccess {
 		return fmt.Errorf("DA layer submission failed: %s", res.Message)

config/config.go

@@ -32,8 +32,13 @@ type NodeConfig struct {
 
 // BlockManagerConfig consists of all parameters required by BlockManagerConfig
 type BlockManagerConfig struct {
-	BlockTime   time.Duration `mapstructure:"block_time"`
-	NamespaceID [8]byte       `mapstructure:"namespace_id"`
+	// BlockTime defines how often new blocks are produced
+	BlockTime time.Duration `mapstructure:"block_time"`
+	// DABlockTime informs about block time of underlying data availability layer
+	DABlockTime time.Duration `mapstructure:"da_block_time"`
+	// DAStartHeight allows skipping first DAStartHeight-1 blocks when querying for blocks.
+	DAStartHeight uint64  `mapstructure:"da_start_height"`
+	NamespaceID   [8]byte `mapstructure:"namespace_id"`
 }
 
 func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {