Merge pull request #55 from datachainlab/feature/v0.5.13-op

Support the new interface of `SetupHeadersForUpdate`, which returns chan

Author: siburu

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/ethereum/go-ethereum v1.14.12
 	github.com/gogo/protobuf v1.3.2
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
-	github.com/hyperledger-labs/yui-relayer v0.5.15
+	github.com/hyperledger-labs/yui-relayer v0.5.16
 	github.com/oasisprotocol/oasis-core/go v0.2403.1
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/viper v1.19.0

go.sum

@@ -736,8 +736,8 @@ github.com/huandu/go-assert v1.1.5/go.mod h1:yOLvuqZwmcHIC5rIzrBhT7D3Q9c3GFnd0Jr
 github.com/huandu/skiplist v1.2.0 h1:gox56QD77HzSC0w+Ws3MH3iie755GBJU1OER3h5VsYw=
 github.com/huandu/skiplist v1.2.0/go.mod h1:7v3iFjLcSAzO4fN5B8dvebvo/qsfumiLiDXMrPiHF9w=
 github.com/hudl/fargo v1.3.0/go.mod h1:y3CKSmjA+wD2gak7sUSXTAoopbhU08POFhmITJgmKTg=
-github.com/hyperledger-labs/yui-relayer v0.5.15 h1:kNJeUV9NJx51Txe9Y4ud6WNWQPMnWXvsReTBjQfp1jg=
-github.com/hyperledger-labs/yui-relayer v0.5.15/go.mod h1:fyY5rY+lqyAAlr1AsWXsB22Qj5phbw/oszgZBL9UeEM=
+github.com/hyperledger-labs/yui-relayer v0.5.16 h1:NauHZ08JsfIvkzopafR/5/TbRCAH7JSUgqm7YrUEQ/4=
+github.com/hyperledger-labs/yui-relayer v0.5.16/go.mod h1:tGNMIE1y4cGNyRRKhR9rfBtCGLVjgejf3EIXouvdDQc=
 github.com/iancoleman/strcase v0.3.0 h1:nTXanmYxhfFAMjZL34Ov6gkzEsSJZ5DbhxWjvSASxEI=
 github.com/iancoleman/strcase v0.3.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=

relay/lcp.go

@@ -397,26 +397,28 @@ func (pr *Prover) updateELC(ctx context.Context, elcClientID string, includeStat
 
 	// 2. query the header from the upstream chain
 
-	headers, err := pr.originProver.SetupHeadersForUpdate(ctx, NewLCPQuerier(pr.lcpServiceClient, elcClientID), latestHeader)
+	headerStream, err := pr.originProver.SetupHeadersForUpdate(ctx, NewLCPQuerier(pr.lcpServiceClient, elcClientID), latestHeader)
 	if err != nil {
 		return nil, err
 	}
-	if len(headers) == 0 {
-		return nil, nil
-	}
 
 	// 3. send a request that contains a header from 2 to update the client in ELC
 	var responses []*elc.MsgUpdateClientResponse
-	for _, header := range headers {
-		anyHeader, err := clienttypes.PackClientMessage(header)
+	i := 0
+	for h := range headerStream {
+		if h.Error != nil {
+			return nil, fmt.Errorf("failed to setup a header for update: i=%v %w", i, h.Error)
+		}
+		anyHeader, err := clienttypes.PackClientMessage(h.Header)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to pack header: i=%v header=%v %w", i, h.Header, err)
 		}
 		res, err := updateClient(ctx, pr.config.GetMaxChunkSizeForUpdateClient(), pr.lcpServiceClient, anyHeader, elcClientID, includeState, pr.activeEnclaveKey.GetEnclaveKeyAddress().Bytes())
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to update ELC: i=%v elc_client_id=%v %w", i, pr.config.ElcClientId, err)
 		}
 		responses = append(responses, res)
+		i += 1
 	}
 
 	return responses, nil

relay/prover.go

@@ -172,26 +172,27 @@ func (pr *Prover) GetLatestFinalizedHeader(ctx context.Context) (core.Header, er
 // SetupHeadersForUpdate returns the finalized header and any intermediate headers needed to apply it to the client on the counterparty chain
 // The order of the returned header slice should be as: [<intermediate headers>..., <update header>]
 // if the header slice's length == nil and err == nil, the relayer should skip the update-client
-func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, dstChain core.FinalityAwareChain, latestFinalizedHeader core.Header) ([]core.Header, error) {
+func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, dstChain core.FinalityAwareChain, latestFinalizedHeader core.Header) (<-chan *core.HeaderOrError, error) {
 	if err := pr.UpdateEKIIfNeeded(ctx, dstChain); err != nil {
 		return nil, err
 	}
 
-	headers, err := pr.originProver.SetupHeadersForUpdate(ctx, dstChain, latestFinalizedHeader)
+	headerStream, err := pr.originProver.SetupHeadersForUpdate(ctx, dstChain, latestFinalizedHeader)
 	if err != nil {
 		return nil, fmt.Errorf("failed to setup headers for update: header=%v %w", latestFinalizedHeader, err)
 	}
-	if len(headers) == 0 {
-		return nil, nil
-	}
 	var (
 		messages   [][]byte
 		signatures [][]byte
 	)
-	for i, h := range headers {
-		anyHeader, err := clienttypes.PackClientMessage(h)
+	i := 0
+	for h := range headerStream {
+		if h.Error != nil {
+			return nil, fmt.Errorf("failed to setup a header for update: i=%v %w", i, h.Error)
+		}
+		anyHeader, err := clienttypes.PackClientMessage(h.Header)
 		if err != nil {
-			return nil, fmt.Errorf("failed to pack header: i=%v header=%v %w", i, h, err)
+			return nil, fmt.Errorf("failed to pack header: i=%v header=%v %w", i, h.Header, err)
 		}
 		res, err := updateClient(ctx, pr.config.GetMaxChunkSizeForUpdateClient(), pr.lcpServiceClient, anyHeader, pr.config.ElcClientId, false, pr.activeEnclaveKey.GetEnclaveKeyAddress().Bytes())
 		if err != nil {
@@ -203,6 +204,10 @@ func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, dstChain core.Final
 		}
 		messages = append(messages, res.Message)
 		signatures = append(signatures, res.Signature)
+		i++
+	}
+	if i == 0 {
+		return core.MakeHeaderStream(), nil
 	}
 
 	var updates []core.Header
@@ -223,7 +228,7 @@ func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, dstChain core.Final
 			})
 		}
 	}
-	return updates, nil
+	return core.MakeHeaderStream(updates...), nil
 }
 
 type MessageAggregator func(ctx context.Context, in *elc.MsgAggregateMessages, opts ...grpc.CallOption) (*elc.MsgAggregateMessagesResponse, error)