Merge branch 'main' into retry_logic_for_core_endpoint

core/eds.go

@@ -24,8 +24,8 @@ import (
 // extendBlock extends the given block data, returning the resulting
 // ExtendedDataSquare (EDS). If there are no transactions in the block,
 // nil is returned in place of the eds.
-func extendBlock(data types.Data, appVersion uint64, options ...nmt.Option) (*rsmt2d.ExtendedDataSquare, error) {
-	if app.IsEmptyBlockRef(&data, appVersion) {
+func extendBlock(data *types.Data, appVersion uint64, options ...nmt.Option) (*rsmt2d.ExtendedDataSquare, error) {
+	if app.IsEmptyBlockRef(data, appVersion) {
 		return share.EmptyEDS(), nil
 	}
 

core/eds_test.go

@@ -22,7 +22,7 @@ func TestTrulyEmptySquare(t *testing.T) {
 		SquareSize: 1,
 	}
 
-	eds, err := extendBlock(data, appconsts.LatestVersion)
+	eds, err := extendBlock(&data, appconsts.LatestVersion)
 	require.NoError(t, err)
 	require.True(t, eds.Equals(share.EmptyEDS()))
 }
@@ -38,7 +38,7 @@ func TestEmptySquareWithZeroTxs(t *testing.T) {
 		Txs: []types.Tx{},
 	}
 
-	eds, err := extendBlock(data, appconsts.LatestVersion)
+	eds, err := extendBlock(&data, appconsts.LatestVersion)
 	require.NoError(t, err)
 	require.True(t, eds.Equals(share.EmptyEDS()))
 

core/exchange.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/tendermint/tendermint/types"
 	"golang.org/x/sync/errgroup"
 
 	libhead "github.com/celestiaorg/go-header"
@@ -62,8 +61,7 @@ func NewExchange(
 
 func (ce *Exchange) GetByHeight(ctx context.Context, height uint64) (*header.ExtendedHeader, error) {
 	log.Debugw("requesting header", "height", height)
-	intHeight := int64(height)
-	return ce.getExtendedHeaderByHeight(ctx, &intHeight)
+	return ce.getExtendedHeaderByHeight(ctx, int64(height))
 }
 
 func (ce *Exchange) GetRangeByHeight(
@@ -129,12 +127,12 @@ func (ce *Exchange) Get(ctx context.Context, hash libhead.Hash) (*header.Extende
 		return nil, fmt.Errorf("fetching block by hash %s: %w", hash.String(), err)
 	}
 
-	comm, vals, err := ce.fetcher.GetBlockInfo(ctx, &block.Height)
+	comm, vals, err := ce.fetcher.GetBlockInfo(ctx, block.Height)
 	if err != nil {
 		return nil, fmt.Errorf("fetching block info for height %d: %w", &block.Height, err)
 	}
 
-	eds, err := extendBlock(block.Data, block.Header.Version.App)
+	eds, err := extendBlock(&block.Data, block.Header.Version.App)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", &block.Height, err)
 	}
@@ -162,29 +160,22 @@ func (ce *Exchange) Head(
 	_ ...libhead.HeadOption[*header.ExtendedHeader],
 ) (*header.ExtendedHeader, error) {
 	log.Debug("requesting head")
-	return ce.getExtendedHeaderByHeight(ctx, nil)
+	return ce.getExtendedHeaderByHeight(ctx, 0)
 }
 
-func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height *int64) (*header.ExtendedHeader, error) {
+func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height int64) (*header.ExtendedHeader, error) {
 	b, err := ce.fetcher.GetSignedBlock(ctx, height)
 	if err != nil {
-		if height == nil {
-			return nil, fmt.Errorf("fetching signed block for head from core: %w", err)
-		}
-		return nil, fmt.Errorf("fetching signed block at height %d from core: %w", *height, err)
+		return nil, fmt.Errorf("fetching signed block at height %d from core: %w", height, err)
 	}
 	log.Debugw("fetched signed block from core", "height", b.Header.Height)
 
 	eds, err := extendBlock(b.Data, b.Header.Version.App)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", b.Header.Height, err)
 	}
-
-	// TODO(@Wondertan): This is a hack to deref Data, allowing GC to pick it up.
-	//  The better footgun-less solution is to change core.ResultSignedBlock fields to be pointers instead of values.
-	b.Data = types.Data{}
-
-	eh, err := ce.construct(&b.Header, &b.Commit, &b.ValidatorSet, eds)
+	// create extended header
+	eh, err := ce.construct(b.Header, b.Commit, b.ValidatorSet, eds)
 	if err != nil {
 		panic(fmt.Errorf("constructing extended header for height %d: %w", b.Header.Height, err))
 	}

core/exchange_test.go

@@ -3,6 +3,7 @@ package core
 import (
 	"bytes"
 	"context"
+	"net"
 	"testing"
 	"time"
 
@@ -34,7 +35,7 @@ func TestCoreExchange_RequestHeaders(t *testing.T) {
 
 	// initialize store with genesis block
 	genHeight := int64(1)
-	genBlock, err := fetcher.GetBlock(ctx, &genHeight)
+	genBlock, err := fetcher.GetBlock(ctx, genHeight)
 	require.NoError(t, err)
 	genHeader, err := ce.Get(ctx, genBlock.Header.Hash().Bytes())
 	require.NoError(t, err)
@@ -61,6 +62,7 @@ func TestCoreExchange_RequestHeaders(t *testing.T) {
 		require.NoError(t, err)
 		assert.True(t, has)
 	}
+	require.NoError(t, fetcher.Stop(ctx))
 }
 
 // TestExchange_DoNotStoreHistoric tests that the CoreExchange will not
@@ -87,7 +89,7 @@ func TestExchange_DoNotStoreHistoric(t *testing.T) {
 
 	// initialize store with genesis block
 	genHeight := int64(1)
-	genBlock, err := fetcher.GetBlock(ctx, &genHeight)
+	genBlock, err := fetcher.GetBlock(ctx, genHeight)
 	require.NoError(t, err)
 	genHeader, err := ce.Get(ctx, genBlock.Header.Hash().Bytes())
 	require.NoError(t, err)
@@ -136,7 +138,7 @@ func TestExchange_StoreHistoricIfArchival(t *testing.T) {
 
 	// initialize store with genesis block
 	genHeight := int64(1)
-	genBlock, err := fetcher.GetBlock(ctx, &genHeight)
+	genBlock, err := fetcher.GetBlock(ctx, genHeight)
 	require.NoError(t, err)
 	genHeader, err := ce.Get(ctx, genBlock.Header.Hash().Bytes())
 	require.NoError(t, err)
@@ -166,7 +168,12 @@ func createCoreFetcher(t *testing.T, cfg *testnode.Config) (*BlockFetcher, testn
 	// flakiness with accessing account state)
 	_, err := cctx.WaitForHeightWithTimeout(2, time.Second*2) // TODO @renaynay: configure?
 	require.NoError(t, err)
-	return NewBlockFetcher(cctx.Client), cctx
+	host, port, err := net.SplitHostPort(cctx.GRPCClient.Target())
+	require.NoError(t, err)
+	client := newTestClient(t, host, port)
+	fetcher, err := NewBlockFetcher(client)
+	require.NoError(t, err)
+	return fetcher, cctx
 }
 
 // fillBlocks fills blocks until the context is canceled.
@@ -202,7 +209,7 @@ func generateNonEmptyBlocks(
 	sub, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 	defer func() {
-		err = fetcher.UnsubscribeNewBlockEvent(ctx)
+		err = fetcher.Stop(ctx)
 		require.NoError(t, err)
 	}()