refactor: revert to use BlockApi and single gRPC connection

.github/workflows/go-ci.yml

@@ -145,7 +145,7 @@ jobs:
 
   integration_test:
     name: Integration Tests
-#    needs: [go_mod_tidy_check, lint, lint-imports, test_coverage] # TODO(chatton): re-enable dependency on lint
+    needs: [go_mod_tidy_check, lint, lint-imports, test_coverage]
     uses: ./.github/workflows/integration-tests.yml
     with:
       go-version: ${{ inputs.go-version }}

api/rpc_test.go

@@ -2,13 +2,13 @@ package api
 
 import (
 	"context"
-	"cosmossdk.io/math"
 	"encoding/json"
 	"reflect"
 	"strconv"
 	"testing"
 	"time"
 
+	"cosmossdk.io/math"
 	"github.com/cristalhq/jwt/v5"
 	"github.com/golang/mock/gomock"
 	"github.com/libp2p/go-libp2p/core/network"

core/eds.go

@@ -3,10 +3,10 @@ package core
 import (
 	"context"
 	"fmt"
-	coretypes "github.com/cometbft/cometbft/types"
 	"time"
 
-	"github.com/celestiaorg/celestia-app/v4/app"
+	coretypes "github.com/cometbft/cometbft/types"
+
 	"github.com/celestiaorg/celestia-app/v4/pkg/appconsts"
 	"github.com/celestiaorg/celestia-app/v4/pkg/wrapper"
 	libsquare "github.com/celestiaorg/go-square/v2"
@@ -20,18 +20,24 @@ import (
 	"github.com/celestiaorg/celestia-node/store"
 )
 
+// isEmptyBlockRef returns true if the application considers the given block data
+// empty at a given version.
+func isEmptyBlockRef(data *coretypes.Data) bool {
+	return len(data.Txs) == 0
+}
+
 // 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 *coretypes.Data, appVersion uint64, options ...nmt.Option) (*rsmt2d.ExtendedDataSquare, error) {
-	if app.IsEmptyBlockRef(data, appVersion) {
+func extendBlock(data *coretypes.Data, options ...nmt.Option) (*rsmt2d.ExtendedDataSquare, error) {
+	if isEmptyBlockRef(data) {
 		return share.EmptyEDS(), nil
 	}
 
 	// Construct the data square from the block's transactions
 	square, err := libsquare.Construct(
 		data.Txs.ToSliceOfBytes(),
-		appconsts.DefaultSquareSizeUpperBound,
+		appconsts.SquareSizeUpperBound,
 		appconsts.SubtreeRootThreshold,
 	)
 	if err != nil {

core/eds_test.go

@@ -4,26 +4,22 @@ import (
 	"testing"
 
 	"github.com/cometbft/cometbft/types"
-	coretypes "github.com/cometbft/cometbft/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/celestiaorg/celestia-app/v4/app"
-	"github.com/celestiaorg/celestia-app/v4/pkg/appconsts"
-
 	"github.com/celestiaorg/celestia-node/share"
 )
 
 // TestTrulyEmptySquare ensures that a truly empty square (square size 1 and no
 // txs) will be recognized as empty and return nil from `extendBlock` so that
 // we do not redundantly store empty EDSes.
 func TestTrulyEmptySquare(t *testing.T) {
-	data := coretypes.Data{
+	data := types.Data{
 		Txs:        []types.Tx{},
 		SquareSize: 1,
 	}
 
-	eds, err := extendBlock(&data, appconsts.LatestVersion)
+	eds, err := extendBlock(&data)
 	require.NoError(t, err)
 	require.True(t, eds.Equals(share.EmptyEDS()))
 }
@@ -35,17 +31,18 @@ func TestTrulyEmptySquare(t *testing.T) {
 // square size do not allow for empty block data. However, should that ever
 // occur, we need to ensure that the correct data root is generated.
 func TestEmptySquareWithZeroTxs(t *testing.T) {
-	data := coretypes.Data{
-		Txs: []coretypes.Tx{},
+	data := types.Data{
+		Txs: []types.Tx{},
 	}
 
-	eds, err := extendBlock(&data, appconsts.LatestVersion)
+	eds, err := extendBlock(&data)
 	require.NoError(t, err)
 	require.True(t, eds.Equals(share.EmptyEDS()))
 
 	// force extend the square using an empty block and compare with the min DAH
-	eds, err = app.ExtendBlock(data)
-	require.NoError(t, err)
+	// TODO(chatton): app.ExtendBlock was removed, can we delete?
+	// eds, err = app.ExtendBlock(data)
+	// require.NoError(t, err)
 
 	roots, err := share.NewAxisRoots(eds)
 	require.NoError(t, err)

core/exchange.go

@@ -132,7 +132,7 @@ func (ce *Exchange) Get(ctx context.Context, hash libhead.Hash) (*header.Extende
 		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)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", &block.Height, err)
 	}
@@ -170,7 +170,7 @@ func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height int64)
 	}
 	log.Debugw("fetched signed block from core", "height", b.Header.Height)
 
-	eds, err := extendBlock(b.Data, b.Header.Version.App)
+	eds, err := extendBlock(b.Data)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", b.Header.Height, err)
 	}

core/exchange_test.go

@@ -3,7 +3,7 @@ package core
 import (
 	"bytes"
 	"context"
-	"github.com/celestiaorg/celestia-node/internal"
+	"net"
 	"testing"
 	"time"
 
@@ -166,7 +166,7 @@ func TestExchange_StoreHistoricIfArchival(t *testing.T) {
 		assert.True(t, has)
 
 		// ensure .q4 file was not stored if not IsEmptyEDS
-		// TODO(chatton): verify if this is the correct behaviour. Does the added WaitForHeight
+		// TODO(chatton): verify if this is the correct behavior. Does the added WaitForHeight
 		// make it so there are some headers that are not empty and so a different code path is followed?
 		has, err = store.HasQ4ByHash(ctx, h.DAH.Hash())
 		require.NoError(t, err)
@@ -180,10 +180,9 @@ 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)
-
-	client, err := internal.NewCoreConn(cfg.TmConfig.RPC.GRPCListenAddress)
+	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

core/fetcher.go

@@ -31,13 +31,13 @@ var (
 )
 
 type BlockFetcher struct {
-	client coregrpc.BlockAPIServiceClient
+	client coregrpc.BlockAPIClient
 }
 
 // NewBlockFetcher returns a new `BlockFetcher`.
 func NewBlockFetcher(conn *grpc.ClientConn) (*BlockFetcher, error) {
 	return &BlockFetcher{
-		client: coregrpc.NewBlockAPIServiceClient(conn),
+		client: coregrpc.NewBlockAPIClient(conn),
 	}, nil
 }
 
@@ -179,7 +179,7 @@ func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (chan types.E
 func (f *BlockFetcher) receive(
 	ctx context.Context,
 	signedBlockCh chan types.EventDataSignedBlock,
-	subscription coregrpc.BlockAPIService_SubscribeNewHeightsClient,
+	subscription coregrpc.BlockAPI_SubscribeNewHeightsClient,
 ) error {
 	log.Debug("fetcher: started listening for new blocks")
 	for {
@@ -221,7 +221,7 @@ func (f *BlockFetcher) IsSyncing(ctx context.Context) (bool, error) {
 	return resp.SyncInfo.CatchingUp, nil
 }
 
-func receiveBlockByHeight(streamer coregrpc.BlockAPIService_BlockByHeightClient) (
+func receiveBlockByHeight(streamer coregrpc.BlockAPI_BlockByHeightClient) (
 	*SignedBlock,
 	error,
 ) {
@@ -264,7 +264,7 @@ func receiveBlockByHeight(streamer coregrpc.BlockAPIService_BlockByHeightClient)
 	}, nil
 }
 
-func receiveBlockByHash(streamer coregrpc.BlockAPIService_BlockByHashClient) (*types.Block, error) {
+func receiveBlockByHash(streamer coregrpc.BlockAPI_BlockByHashClient) (*types.Block, error) {
 	parts := make([]*tmproto.Part, 0)
 	isLast := false
 	for !isLast {

core/fetcher_no_race_test.go

@@ -4,7 +4,7 @@ package core
 
 import (
 	"context"
-	"github.com/celestiaorg/celestia-node/internal"
+	"net"
 	"testing"
 	"time"
 
@@ -19,10 +19,10 @@ func TestBlockFetcherHeaderValues(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
 	t.Cleanup(cancel)
 
-	cfg := DefaultTestConfig()
-	StartTestNodeWithConfig(t, cfg)
-	client, err := internal.NewCoreConn(cfg.TmConfig.RPC.GRPCListenAddress)
+	node := StartTestNode(t)
+	host, port, err := net.SplitHostPort(node.GRPCClient.Target())
 	require.NoError(t, err)
+	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
 	newBlockChan, err := fetcher.SubscribeNewBlockEvent(ctx)

core/fetcher_test.go

@@ -2,7 +2,7 @@ package core
 
 import (
 	"context"
-	"github.com/celestiaorg/celestia-node/internal"
+	"net"
 	"testing"
 	"time"
 
@@ -14,9 +14,9 @@ func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
 	t.Cleanup(cancel)
 
-	cfg := DefaultTestConfig()
-	StartTestNodeWithConfig(t, cfg)
-	client, err := internal.NewCoreConn(cfg.TmConfig.RPC.GRPCListenAddress)
+	host, port, err := net.SplitHostPort(StartTestNode(t).GRPCClient.Target())
+	require.NoError(t, err)
+	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
 	// generate some blocks
@@ -49,8 +49,9 @@ func TestFetcher_Resubscription(t *testing.T) {
 	cfg := DefaultTestConfig()
 	tn := NewNetwork(t, cfg)
 	require.NoError(t, tn.Start())
-	client, err := internal.NewCoreConn(cfg.TmConfig.RPC.GRPCListenAddress)
+	host, port, err := net.SplitHostPort(tn.GRPCClient.Target())
 	require.NoError(t, err)
+	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
 
@@ -84,15 +85,6 @@ func TestFetcher_Resubscription(t *testing.T) {
 	// on the same address and listen for the new blocks
 	tn = NewNetwork(t, cfg)
 	require.NoError(t, tn.Start())
-
-	// TODO(chatton): verify the test is still testing what it was originally testing.
-	client, err = internal.NewCoreConn(cfg.TmConfig.RPC.GRPCListenAddress)
-	require.NoError(t, err)
-	fetcher, err = NewBlockFetcher(client)
-	require.NoError(t, err)
-	newBlockChan, err = fetcher.SubscribeNewBlockEvent(ctx)
-	require.NoError(t, err)
-
 	select {
 	case newBlockFromChan := <-newBlockChan:
 		h := newBlockFromChan.Header.Height

core/header_test.go

@@ -3,7 +3,7 @@ package core
 import (
 	"context"
 	"fmt"
-	"github.com/celestiaorg/celestia-node/internal"
+	"net"
 	"testing"
 
 	"github.com/cometbft/cometbft/libs/rand"
@@ -21,12 +21,9 @@ func TestMakeExtendedHeaderForEmptyBlock(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	cfg := DefaultTestConfig()
-	StartTestNodeWithConfig(t, cfg)
-
-	client, err := internal.NewCoreConn(cfg.TmConfig.RPC.GRPCListenAddress)
-
+	host, port, err := net.SplitHostPort(StartTestNode(t).GRPCClient.Target())
 	require.NoError(t, err)
+	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
 	sub, err := fetcher.SubscribeNewBlockEvent(ctx)
@@ -40,7 +37,7 @@ func TestMakeExtendedHeaderForEmptyBlock(t *testing.T) {
 	comm, val, err := fetcher.GetBlockInfo(ctx, height)
 	require.NoError(t, err)
 
-	eds, err := extendBlock(b.Data, b.Header.Version.App)
+	eds, err := extendBlock(b.Data)
 	require.NoError(t, err)
 
 	headerExt, err := header.MakeExtendedHeader(b.Header, comm, val, eds)

core/listener.go

@@ -177,7 +177,7 @@ func (cl *Listener) handleNewSignedBlock(ctx context.Context, b types.EventDataS
 		attribute.Int64("height", b.Header.Height),
 	)
 
-	eds, err := extendBlock(&b.Data, b.Header.Version.App)
+	eds, err := extendBlock(&b.Data)
 	if err != nil {
 		return fmt.Errorf("extending block data: %w", err)
 	}