Merge pull request #166 from hyperledger-labs/feature/v0.5.13-op

Change the return value type of `Prover::SetupHeadersForUpdate` to support stream processing of huge headers

Author: siburu

chains/tendermint/prover.go

@@ -91,7 +91,7 @@ func (pr *Prover) CreateInitialLightClientState(ctx context.Context, height ibce
 }
 
 // SetupHeadersForUpdate returns the finalized header and any intermediate headers needed to apply it to the client on the counterpaty chain
-func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.FinalityAwareChain, latestFinalizedHeader core.Header) ([]core.Header, error) {
+func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.FinalityAwareChain, latestFinalizedHeader core.Header) (<-chan *core.HeaderOrError, error) {
 	self := pr.chain
 	// make copy of header stored in mop
 	tmp := latestFinalizedHeader.(*tmclient.Header)
@@ -124,7 +124,7 @@ func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.F
 
 	// inject TrustedValidators into header
 	h.TrustedValidators = valSet
-	return []core.Header{&h}, nil
+	return core.MakeHeaderStream(&h), nil
 }
 
 // GetLatestFinalizedHeader returns the latest finalized header

core/channel-upgrade.go

@@ -228,7 +228,7 @@ func CancelChannelUpgrade(ctx context.Context, chain, cp *ProvableChain, settlem
 			return false, nil
 		}
 
-		cpHeaders, err := cp.SetupHeadersForUpdate(ctx, chain, sh.GetLatestFinalizedHeader(cp.ChainID()))
+		cpHeaders, err := SetupHeadersForUpdateSync(cp, ctx, chain, sh.GetLatestFinalizedHeader(cp.ChainID()))
 		if err != nil {
 			logger.ErrorContext(ctx, "failed to set up headers for LC update", err)
 			return false, err

core/header-stream.go

@@ -0,0 +1,33 @@
+package core
+
+import "context"
+
+func SetupHeadersForUpdateSync(prover LightClient, ctx context.Context, counterparty FinalityAwareChain, latestFinalizedHeader Header) ([]Header, error) {
+	ctxForSHFU, cancel := context.WithCancel(ctx)
+	defer cancel()
+	headerStream, err := prover.SetupHeadersForUpdate(ctxForSHFU, counterparty, latestFinalizedHeader)
+	if err != nil {
+		return nil, err
+	}
+
+	var ret []Header
+	for h := range headerStream {
+		if h.Error != nil {
+			return nil, h.Error
+		}
+		ret = append(ret, h.Header)
+	}
+	return ret, nil
+}
+
+func MakeHeaderStream(headers ...Header) <-chan *HeaderOrError {
+	ch := make(chan *HeaderOrError, len(headers))
+	for _, h := range headers {
+		ch <- &HeaderOrError{
+			Header: h,
+			Error:  nil,
+		}
+	}
+	close(ch)
+	return ch
+}

core/headers.go

@@ -135,7 +135,7 @@ func (sh syncHeaders) SetupHeadersForUpdate(ctx context.Context, src, dst ChainL
 		logger.ErrorContext(ctx, "error ensuring different chains", err)
 		return nil, err
 	}
-	return src.SetupHeadersForUpdate(ctx, dst, sh.GetLatestFinalizedHeader(src.ChainID()))
+	return SetupHeadersForUpdateSync(src, ctx, dst, sh.GetLatestFinalizedHeader(src.ChainID()))
 }
 
 // SetupBothHeadersForUpdate returns both `src` and `dst` chain's headers to update the clients on each chain

core/provers.go

@@ -34,6 +34,11 @@ type StateProver interface {
 	ProveHostConsensusState(ctx QueryContext, height exported.Height, consensusState exported.ConsensusState) (proof []byte, err error)
 }
 
+type HeaderOrError struct {
+	Header Header
+	Error  error
+}
+
 // LightClient provides functions for creating and updating on-chain light clients on the counterparty chain
 type LightClient interface {
 	FinalityAware
@@ -44,9 +49,11 @@ type LightClient interface {
 	CreateInitialLightClientState(ctx context.Context, height exported.Height) (exported.ClientState, exported.ConsensusState, error)
 
 	// SetupHeadersForUpdate returns the finalized header and any intermediate headers needed to apply it to the client on the counterpaty chain
-	// The order of the returned header slice should be as: [<intermediate headers>..., <update header>]
-	// if the header slice's length == 0 and err == nil, the relayer should skips the update-client
-	SetupHeadersForUpdate(ctx context.Context, counterparty FinalityAwareChain, latestFinalizedHeader Header) ([]Header, error)
+	// CONTRACT:
+	// 1. The order of the returned header stream should be as: [<intermediate headers>..., <update header>]
+	// 2. If the header stream's length == 0 and err == nil, the relayer should skips the update-client
+	// 3. Goroutines that create the header stream should check `ctx.Done()` and should terminate processing if `ctx` is cancelled by the caller.
+	SetupHeadersForUpdate(ctx context.Context, counterparty FinalityAwareChain, latestFinalizedHeader Header) (<-chan *HeaderOrError, error)
 
 	// CheckRefreshRequired returns if the on-chain light client needs to be updated.
 	// For example, this requirement arises due to the trusting period mechanism.

otelcore/prover.go

@@ -52,7 +52,7 @@ func (p *Prover) CreateInitialLightClientState(ctx context.Context, height ibcex
 	return clientState, consensusState, err
 }
 
-func (p *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.FinalityAwareChain, latestFinalizedHeader core.Header) ([]core.Header, error) {
+func (p *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.FinalityAwareChain, latestFinalizedHeader core.Header) (<-chan *core.HeaderOrError, error) {
 	ctx, span := p.tracer.Start(ctx, "Prover.SetupHeadersForUpdate",
 		core.WithChainAttributes(p.chainID),
 		trace.WithAttributes(attribute.String("counterparty_chain_id", counterparty.ChainID())),

provers/mock/prover.go

@@ -72,8 +72,8 @@ func (pr *Prover) CreateInitialLightClientState(ctx context.Context, height expo
 }
 
 // SetupHeadersForUpdate returns the finalized header and any intermediate headers needed to apply it to the client on the counterparty chain
-func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.FinalityAwareChain, latestFinalizedHeader core.Header) ([]core.Header, error) {
-	return []core.Header{latestFinalizedHeader.(*mocktypes.Header)}, nil
+func (pr *Prover) SetupHeadersForUpdate(ctx context.Context, counterparty core.FinalityAwareChain, latestFinalizedHeader core.Header) (<-chan *core.HeaderOrError, error) {
+	return core.MakeHeaderStream(latestFinalizedHeader.(*mocktypes.Header)), nil
 }
 
 func (pr *Prover) createMockHeader(ctx context.Context, height exported.Height) (core.Header, error) {