Confirm successful Eth header import in parachain (Snowfork#421)

* add OnProcessed callback to extrinsic pool * Expose Eth header ID fields * confirm existence of imported header
longlonghash · Jun 7, 2021 · 30e40ae · 30e40ae
1 parent 4184e95
commit 30e40ae
Show file tree

Hide file tree

Showing 5 changed files with 116 additions and 53 deletions.
diff --git a/relayer/chain/ethereum/header.go b/relayer/chain/ethereum/header.go
@@ -12,6 +12,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/snowfork/ethashproof"
 	"github.com/snowfork/ethashproof/ethash"
+	"github.com/snowfork/go-substrate-rpc-client/v2/scale"
 	types "github.com/snowfork/go-substrate-rpc-client/v2/types"
 	"github.com/snowfork/polkadot-ethereum/relayer/chain"
 )
@@ -21,7 +22,7 @@ type HeaderID struct {
 	Hash   types.H256
 }
 
-type Header struct {
+type headerSCALE struct {
 	ParentHash       types.H256
 	Timestamp        types.U64
 	Number           types.U64
@@ -38,6 +39,33 @@ type Header struct {
 	Seal             []types.Bytes
 }
 
+type Header struct {
+	Fields headerSCALE
+	header *etypes.Header
+}
+
+func (h *Header) Decode(decoder scale.Decoder) error {
+	var fields headerSCALE
+	err := decoder.Decode(&fields)
+	if err != nil {
+		return err
+	}
+
+	h.Fields = fields
+	return nil
+}
+
+func (h Header) Encode(encoder scale.Encoder) error {
+	return encoder.Encode(h.Fields)
+}
+
+func (h *Header) ID() HeaderID {
+	return HeaderID{
+		Number: h.Fields.Number,
+		Hash:   types.NewH256(h.header.Hash().Bytes()),
+	}
+}
+
 type DoubleNodeWithMerkleProof struct {
 	DagNodes [2]types.H512
 	Proof    [][16]byte
@@ -92,20 +120,23 @@ func MakeHeaderData(gethheader *etypes.Header) (*Header, error) {
 	}
 
 	return &Header{
-		ParentHash:       types.NewH256(gethheader.ParentHash.Bytes()),
-		Timestamp:        types.NewU64(gethheader.Time),
-		Number:           types.NewU64(blockNumber),
-		Author:           types.NewH160(gethheader.Coinbase.Bytes()),
-		TransactionsRoot: types.NewH256(gethheader.TxHash.Bytes()),
-		OmmersHash:       types.NewH256(gethheader.UncleHash.Bytes()),
-		ExtraData:        types.NewBytes(gethheader.Extra),
-		StateRoot:        types.NewH256(gethheader.Root.Bytes()),
-		ReceiptsRoot:     types.NewH256(gethheader.ReceiptHash.Bytes()),
-		LogsBloom:        types.NewBytes256(bloomBytes),
-		GasUsed:          types.NewU256(gasUsed),
-		GasLimit:         types.NewU256(gasLimit),
-		Difficulty:       types.NewU256(*gethheader.Difficulty),
-		Seal:             []types.Bytes{mixHashRLP, nonceRLP},
+		Fields: headerSCALE{
+			ParentHash:       types.NewH256(gethheader.ParentHash.Bytes()),
+			Timestamp:        types.NewU64(gethheader.Time),
+			Number:           types.NewU64(blockNumber),
+			Author:           types.NewH160(gethheader.Coinbase.Bytes()),
+			TransactionsRoot: types.NewH256(gethheader.TxHash.Bytes()),
+			OmmersHash:       types.NewH256(gethheader.UncleHash.Bytes()),
+			ExtraData:        types.NewBytes(gethheader.Extra),
+			StateRoot:        types.NewH256(gethheader.Root.Bytes()),
+			ReceiptsRoot:     types.NewH256(gethheader.ReceiptHash.Bytes()),
+			LogsBloom:        types.NewBytes256(bloomBytes),
+			GasUsed:          types.NewU256(gasUsed),
+			GasLimit:         types.NewU256(gasLimit),
+			Difficulty:       types.NewU256(*gethheader.Difficulty),
+			Seal:             []types.Bytes{mixHashRLP, nonceRLP},
+		},
+		header: gethheader,
 	}, nil
 }
 

diff --git a/relayer/chain/ethereum/header_test.go b/relayer/chain/ethereum/header_test.go
@@ -133,7 +133,7 @@ func TestHeader_EncodeDecode11090290(t *testing.T) {
 	if err != nil {
 		panic(err)
 	}
-	assert.Equal(t, *header, decoded, "Decoded Substrate header should match ethereum.Header")
+	assert.Equal(t, header.Fields, decoded.Fields, "Decoded Substrate header should match ethereum.Header")
 }
 
 func TestProof_EncodeDecode(t *testing.T) {

diff --git a/relayer/chain/parachain/outgoing_extrinsics.go b/relayer/chain/parachain/outgoing_extrinsics.go
@@ -35,17 +35,17 @@ func NewExtrinsicPool(eg *errgroup.Group, conn *Connection, log *logrus.Entry) *
 	return &ep
 }
 
-func (ep *ExtrinsicPool) WaitForSubmitAndWatch(ctx context.Context, nonce uint32, ext *types.Extrinsic) {
+func (ep *ExtrinsicPool) WaitForSubmitAndWatch(ctx context.Context, nonce uint32, ext *types.Extrinsic, onProcessed func() error) {
 	select {
 	case ep.watched <- struct{}{}:
 		ep.eg.Go(func() error {
-			return ep.submitAndWatchLoop(ctx, nonce, ext)
+			return ep.submitAndWatchLoop(ctx, nonce, ext, onProcessed)
 		})
 	case <-ctx.Done():
 	}
 }
 
-func (ep *ExtrinsicPool) submitAndWatchLoop(ctx context.Context, nonce uint32, ext *types.Extrinsic) error {
+func (ep *ExtrinsicPool) submitAndWatchLoop(ctx context.Context, nonce uint32, ext *types.Extrinsic, onProcessed func() error) error {
 	sub, err := ep.conn.api.RPC.Author.SubmitAndWatchExtrinsic(*ext)
 	if err != nil {
 		return err
@@ -105,7 +105,7 @@ func (ep *ExtrinsicPool) submitAndWatchLoop(ctx context.Context, nonce uint32, e
 					ep.maxNonce = nonce
 				}
 				<-ep.watched
-				return nil
+				return onProcessed()
 			}
 
 		case err := <-sub.Err():

diff --git a/relayer/cmd/getblock.go b/relayer/cmd/getblock.go
@@ -170,37 +170,37 @@ func printEthBlockForSub(header *gethTypes.Header, format Format) error {
 				hex!("%x").to_vec(),
 			],
 		}`,
-			headerForSub.ParentHash,
+			headerForSub.Fields.ParentHash,
 			header.Time,
-			headerForSub.Number,
-			headerForSub.Author,
-			headerForSub.TransactionsRoot,
-			headerForSub.OmmersHash,
-			headerForSub.ExtraData,
-			headerForSub.StateRoot,
-			headerForSub.ReceiptsRoot,
-			headerForSub.LogsBloom,
-			headerForSub.GasUsed,
-			headerForSub.GasLimit,
-			headerForSub.Difficulty,
-			headerForSub.Seal[0],
-			headerForSub.Seal[1],
+			headerForSub.Fields.Number,
+			headerForSub.Fields.Author,
+			headerForSub.Fields.TransactionsRoot,
+			headerForSub.Fields.OmmersHash,
+			headerForSub.Fields.ExtraData,
+			headerForSub.Fields.StateRoot,
+			headerForSub.Fields.ReceiptsRoot,
+			headerForSub.Fields.LogsBloom,
+			headerForSub.Fields.GasUsed,
+			headerForSub.Fields.GasLimit,
+			headerForSub.Fields.Difficulty,
+			headerForSub.Fields.Seal[0],
+			headerForSub.Fields.Seal[1],
 		)
 		fmt.Println("")
 	} else {
-		extraData, err := json.Marshal(bytesAsArray64(headerForSub.ExtraData))
+		extraData, err := json.Marshal(bytesAsArray64(headerForSub.Fields.ExtraData))
 		if err != nil {
 			return err
 		}
-		logsBloom, err := json.Marshal(headerForSub.LogsBloom)
+		logsBloom, err := json.Marshal(headerForSub.Fields.LogsBloom)
 		if err != nil {
 			return err
 		}
-		seal1, err := json.Marshal(bytesAsArray64(headerForSub.Seal[0]))
+		seal1, err := json.Marshal(bytesAsArray64(headerForSub.Fields.Seal[0]))
 		if err != nil {
 			return err
 		}
-		seal2, err := json.Marshal(bytesAsArray64(headerForSub.Seal[1]))
+		seal2, err := json.Marshal(bytesAsArray64(headerForSub.Fields.Seal[1]))
 		if err != nil {
 			return err
 		}
@@ -225,19 +225,19 @@ func printEthBlockForSub(header *gethTypes.Header, format Format) error {
 				%s
 			]
 		}`,
-			headerForSub.ParentHash.Hex(),
+			headerForSub.Fields.ParentHash.Hex(),
 			header.Time,
-			headerForSub.Number,
-			headerForSub.Author.Hex(),
-			headerForSub.TransactionsRoot.Hex(),
-			headerForSub.OmmersHash.Hex(),
+			headerForSub.Fields.Number,
+			headerForSub.Fields.Author.Hex(),
+			headerForSub.Fields.TransactionsRoot.Hex(),
+			headerForSub.Fields.OmmersHash.Hex(),
 			extraData,
-			headerForSub.StateRoot.Hex(),
-			headerForSub.ReceiptsRoot.Hex(),
+			headerForSub.Fields.StateRoot.Hex(),
+			headerForSub.Fields.ReceiptsRoot.Hex(),
 			logsBloom,
-			headerForSub.GasUsed,
-			headerForSub.GasLimit,
-			headerForSub.Difficulty,
+			headerForSub.Fields.GasUsed,
+			headerForSub.Fields.GasLimit,
+			headerForSub.Fields.Difficulty,
 			seal1,
 			seal2,
 		)

diff --git a/relayer/workers/ethrelayer/parachain-writer.go b/relayer/workers/ethrelayer/parachain-writer.go
@@ -97,6 +97,24 @@ func (wr *ParachainWriter) queryAccountNonce() (uint32, error) {
 	return uint32(accountInfo.Nonce), nil
 }
 
+func (wr *ParachainWriter) queryImportedHeaderExists(hash types.H256) (bool, error) {
+	key, err := types.CreateStorageKey(wr.conn.GetMetadata(), "VerifierLightclient", "Headers", hash[:], nil)
+	if err != nil {
+		return false, err
+	}
+
+	var headerOption types.OptionBytes
+	ok, err := wr.conn.GetAPI().RPC.State.GetStorageLatest(key, &headerOption)
+	if err != nil {
+		return false, err
+	}
+	if !ok {
+		return false, fmt.Errorf("Unable to query header for hash %s", hash.Hex())
+	}
+
+	return headerOption.IsSome(), nil
+}
+
 func (wr *ParachainWriter) writeLoop(ctx context.Context) error {
 	for {
 		select {
@@ -107,25 +125,26 @@ func (wr *ParachainWriter) writeLoop(ctx context.Context) error {
 				return nil
 			}
 
+			header := payload.Header.HeaderData.(ethereum.Header)
 			err := wr.WritePayload(ctx, &payload)
 			if err != nil {
 				wr.log.WithError(err).WithFields(logrus.Fields{
-					"blockNumber":  payload.Header.HeaderData.(ethereum.Header).Number,
+					"blockNumber":  header.Fields.Number,
 					"messageCount": len(payload.Messages),
 				}).Error("Failure submitting header and messages to Substrate")
 				return err
 			}
 
 			wr.log.WithFields(logrus.Fields{
-				"blockNumber":  payload.Header.HeaderData.(ethereum.Header).Number,
+				"blockNumber":  header.Fields.Number,
 				"messageCount": len(payload.Messages),
 			}).Info("Submitted header and messages to Substrate")
 		}
 	}
 }
 
 // Write submits a transaction to the chain
-func (wr *ParachainWriter) write(ctx context.Context, c types.Call) error {
+func (wr *ParachainWriter) write(ctx context.Context, c types.Call, onProcessed func() error) error {
 	ext := types.NewExtrinsic(c)
 
 	latestHash, err := wr.conn.GetAPI().RPC.Chain.GetFinalizedHead()
@@ -161,7 +180,7 @@ func (wr *ParachainWriter) write(ctx context.Context, c types.Call) error {
 		return err
 	}
 
-	wr.pool.WaitForSubmitAndWatch(ctx, wr.nonce, &extI)
+	wr.pool.WaitForSubmitAndWatch(ctx, wr.nonce, &extI, onProcessed)
 
 	wr.nonce = wr.nonce + 1
 
@@ -189,7 +208,20 @@ func (wr *ParachainWriter) WritePayload(ctx context.Context, payload *ParachainP
 		return err
 	}
 
-	return wr.write(ctx, call)
+	onProcessed := func() error {
+		// Confirm that the header import was successful
+		header := payload.Header.HeaderData.(ethereum.Header)
+		hash := header.ID().Hash
+		imported, err := wr.queryImportedHeaderExists(hash)
+		if err != nil {
+			return err
+		}
+		if !imported {
+			return fmt.Errorf("Header import failed for header %s", hash.Hex())
+		}
+		return nil
+	}
+	return wr.write(ctx, call, onProcessed)
 }
 
 func (wr *ParachainWriter) makeMessageSubmitCall(ctx context.Context, msg *chain.EthereumOutboundMessage) (types.Call, error) {