Ensure successful block submission even when sequencer crashes

.github/workflows/docker-build-test-ethermint.yml

@@ -37,6 +37,7 @@ jobs:
         with:
           repository: celestiaorg/ethermint
           path: ethermint
+          ref: optimint-v0.3.0-rebase
       - name: Move Stuff and Tidy
         run: |
           ls
@@ -84,11 +85,13 @@ jobs:
       with:
         repository: celestiaorg/ephemeral-cluster
         path: ephemeral-cluster
+        ref: jbowen93/mess
     - name: Clone ethermint repo
       uses: actions/checkout@v3
       with:
         repository: celestiaorg/ethermint
         path: ethermint
+        ref: optimint-v0.3.0-rebase
     - name: Set up Go
       uses: actions/setup-go@v2
       with:
@@ -105,13 +108,9 @@ jobs:
         docker ps -a
         echo "------------- docker logs core0 -------------"
         docker logs core0
-        echo "------------- docker logs bridge0 -------------"
-        docker logs bridge0
-        echo "------------- docker logs light0 -------------"
-        docker logs light0
         echo "------------- docker logs dalc0 -------------"
         docker logs dalc0
-        echo "------------- docker logs evmos0 -------------"
+        echo "------------- docker logs ethermint0 -------------"
         docker logs ethermint0
         cd ethermint
         echo "------------- go test ./tests/rpc/... -------------"

block/manager.go

@@ -197,6 +197,7 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 			daHeight := blockEvent.daHeight
 			m.logger.Debug("block body retrieved from DALC",
 				"height", block.Header.Height,
+				"daHeight", daHeight,
 				"hash", block.Hash(),
 			)
 			m.syncCache[block.Header.Height] = block
@@ -205,7 +206,8 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 			b1, ok1 := m.syncCache[currentHeight+1]
 			b2, ok2 := m.syncCache[currentHeight+2]
 			if ok1 && ok2 {
-				newState, responses, _, err := m.executor.ApplyBlock(ctx, m.lastState, b1)
+				m.logger.Info("Syncing block", "height", b1.Header.Height)
+				newState, responses, err := m.executor.ApplyBlock(ctx, m.lastState, b1)
 				if err != nil {
 					m.logger.Error("failed to ApplyBlock", "error", err)
 					continue
@@ -215,7 +217,14 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 					m.logger.Error("failed to save block", "error", err)
 					continue
 				}
-				err = m.store.SaveBlockResponses(block.Header.Height, responses)
+				_, _, err = m.executor.Commit(ctx, newState, b1, responses)
+				if err != nil {
+					m.logger.Error("failed to Commit", "error", err)
+					continue
+				}
+				m.store.SetHeight(b1.Header.Height)
+
+				err = m.store.SaveBlockResponses(b1.Header.Height, responses)
 				if err != nil {
 					m.logger.Error("failed to save block responses", "error", err)
 					continue
@@ -339,63 +348,90 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 		lastHeaderHash = lastBlock.Header.Hash()
 	}
 
-	m.logger.Info("Creating and publishing block", "height", newHeight)
+	var block *types.Block
 
-	block := m.executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, m.lastState)
-	m.logger.Debug("block info", "num_tx", len(block.Data.Txs))
-	newState, responses, _, err := m.executor.ApplyBlock(ctx, m.lastState, block)
-	if err != nil {
-		return err
+	// Check if there's an already stored block at a newer height
+	// If there is use that instead of creating a new block
+	pendingBlock, err := m.store.LoadBlock(newHeight)
+	if err == nil {
+		m.logger.Info("Using pending block", "height", newHeight)
+		block = pendingBlock
+	} else {
+		m.logger.Info("Creating and publishing block", "height", newHeight)
+		block = m.executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, m.lastState)
+		m.logger.Debug("block info", "num_tx", len(block.Data.Txs))
+
+		headerBytes, err := block.Header.MarshalBinary()
+		if err != nil {
+			return err
+		}
+		sign, err := m.proposerKey.Sign(headerBytes)
+		if err != nil {
+			return err
+		}
+		commit := &types.Commit{
+			Height:     block.Header.Height,
+			HeaderHash: block.Header.Hash(),
+			Signatures: []types.Signature{sign},
+		}
+
+		// SaveBlock commits the DB tx
+		err = m.store.SaveBlock(block, commit)
+		if err != nil {
+			return err
+		}
 	}
 
-	headerBytes, err := block.Header.MarshalBinary()
+	// Apply the block but DONT commit
+	newState, responses, err := m.executor.ApplyBlock(ctx, m.lastState, block)
 	if err != nil {
 		return err
 	}
-	sign, err := m.proposerKey.Sign(headerBytes)
+
+	err = m.submitBlockToDA(ctx, block)
 	if err != nil {
+		m.logger.Error("Failed to submit block to DA Layer")
 		return err
 	}
 
-	commit := &types.Commit{
-		Height:     block.Header.Height,
-		HeaderHash: block.Header.Hash(),
-		Signatures: []types.Signature{sign},
-	}
-	err = m.store.SaveBlock(block, commit)
+	// Only update the stored height after successfully submitting to DA layer
+	m.store.SetHeight(block.Header.Height)
+
+	// Commit the new state and block which writes to disk on the proxy app
+	_, _, err = m.executor.Commit(ctx, newState, block, responses)
 	if err != nil {
 		return err
 	}
 
+	// SaveBlockResponses commits the DB tx
 	err = m.store.SaveBlockResponses(block.Header.Height, responses)
 	if err != nil {
 		return err
 	}
 
 	newState.DAHeight = atomic.LoadUint64(&m.daHeight)
+	// After this call m.lastState is the NEW state returned from ApplyBlock
 	m.lastState = newState
+
+	// UpdateState commits the DB tx
 	err = m.store.UpdateState(m.lastState)
 	if err != nil {
 		return err
 	}
 
+	// SaveValidators commits the DB tx
 	err = m.store.SaveValidators(block.Header.Height, m.lastState.Validators)
 	if err != nil {
 		return err
 	}
 
-	err = m.submitBlockToDA(ctx, block)
-	if err != nil {
-		return err
-	}
-
 	m.publishHeader(block)
 
 	return nil
 }
 
 func (m *Manager) submitBlockToDA(ctx context.Context, block *types.Block) error {
-	m.logger.Debug("submitting block to DA layer", "height", block.Header.Height)
+	m.logger.Info("submitting block to DA layer", "height", block.Header.Height)
 
 	submitted := false
 	backoff := initialBackoff
@@ -412,8 +448,7 @@ func (m *Manager) submitBlockToDA(ctx context.Context, block *types.Block) error
 	}
 
 	if !submitted {
-		// TODO(tzdybal): probably this could be handled better
-		panic("Failed to submit block to DA layer!")
+		return fmt.Errorf("Failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
 	}
 
 	m.HeaderOutCh <- &block.Header

go.mod

@@ -44,7 +44,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
 	github.com/dgraph-io/badger/v2 v2.2007.2 // indirect
-	github.com/dgraph-io/ristretto v0.1.0 // indirect
+	github.com/dgraph-io/ristretto v0.1.1-0.20210901224026-c2901dc76889 // indirect
 	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
 	github.com/dustin/go-humanize v1.0.0 // indirect
@@ -57,7 +57,7 @@ require (
 	github.com/go-resty/resty/v2 v2.7.0 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
-	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
+	github.com/golang/glog v1.0.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.3 // indirect
@@ -183,7 +183,7 @@ require (
 	golang.org/x/crypto v0.0.0-20220518034528-6f7dac969898 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.10 // indirect
 	golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df // indirect