Repository: celestiaorg/celestia-node (go)

Issue:
Bridge Node Stuck After Archive RPC Restart
### Celestia Node version

v0.21.3-mocha

### Celestia Consensus Node version

3.3.0-mocha

### OS

22.04.5 LTS (Jammy Jellyfish)

### Steps to reproduce it

Restart the `archive RPC` while the bridge node is running

### Expected result

The bridge node should be able to recover from an archive RPC restart without requiring a manual service restart.

### Actual result

The bridge node became unresponsive. It repeatedly logged fetcher and listener errors. The issue persisted until a manual restart was performed.

### Relevant log output

```Shell
2025-02-05T07:28:54.515Z        INFO    header/store    store/store.go:367      new head        {"height": 4529796, "hash": "4881C54F1993FBC1BEDC574CBBA71A798210B88C17572DBE970B0673E83FC110"}
2025-02-05T07:28:59.391Z        INFO    header/store    store/store.go:367      new head        {"height": 4529797, "hash": "7953D784D1982641504D365BF30BF8AF3ADE797EB722B87996296C87FAB85BF1"}
2025-02-05T07:29:04.389Z        INFO    header/store    store/store.go:367      new head        {"height": 4529798, "hash": "3A798993AD62B133BB73EA63FEBDF29B1D06B21AA650335B9ADAEB05A38CF754"}
2025-02-05T07:29:09.274Z        INFO    header/store    store/store.go:367      new head        {"height": 4529799, "hash": "B6F80329D1DA76F3602C1D5FB91B27C04835E4AA142F6221E7A175A01015E55B"}
2025-02-05T07:29:14.243Z        INFO    header/store    store/store.go:367      new head        {"height": 4529800, "hash": "BCE184F68E3C165FF60C2BB569CA1DB37C78F2356B34039FFD4261557A049DAF"}
2025-02-05T07:29:17.963Z        INFO    bitswap/server/decision decision/engine.go:733  handling wantlist overflow      {"local": "12D3KooWKdtKbb6KHYnnEjxdMgBT3eVKmuvT8ykjgbajcSKz89LP", "from": "12D3KooWDHfTaYvQE1z4DBKRtH5rKnJc5RrBpjrYhM1EFYttyCkt", "wantlistSize": 4093, "overflowSize": 3}
2025-02-05T07:29:19.217Z        INFO    header/store    store/store.go:367      new head        {"height": 4529801, "hash": "ACA69B7C5EFD9F98372B8D1E18255FBD30B77AD7D76F4299C486B3391278E01C"}
2025-02-05T07:29:24.286Z        INFO    header/store    store/store.go:367      new head        {"height": 4529802, "hash": "91B489EC15D4F1D5CFC0D7B1BD8AC8DF37D5A6FA6159836FB9F09A8E2F8C91D0"}
2025-02-05T07:29:29.497Z        INFO    header/store    store/store.go:367      new head        {"height": 4529803, "hash": "3530B64936561099C82A9E3DF4E4BDBF6359BEC8D196C299A9A155755453ED88"}
2025-02-05T07:29:31.733Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:29:31.734Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:29:31.734Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:29:31.734Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:29:31.734Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
...
2025-02-05T07:30:09.864Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:30:09.864Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:30:09.864Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:30:09.864Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
2025-02-05T07:30:09.864Z        ERROR   core    core/fetcher.go:191     fetcher: error receiving new height     {"err": "rpc error: code = Unavailable desc = error reading from server: EOF"}
Suppressed 1363430 messages from celestia-bridge.service
2025-02-05T07:30:39.499Z        ERROR   core    core/listener.go:171    listener: resubscribe   {"err": "already subscribed to new blocks"}
2025-02-05T07:30:44.499Z        ERROR   core    core/listener.go:171    listener: resubscribe   {"err": "already subscribed to new blocks"}
2025-02-05T07:30:47.994Z        INFO    bitswap/server/decision decision/engine.go:733  handling wantlist overflow      {"local": "12D3KooWKdtKbb6KHYnnEjxdMgBT3eVKmuvT8ykjgbajcSKz89LP", "from": "12D3KooWDHfTaYvQE1z4DBKRtH5rKnJc5RrBpjrYhM1EFYttyCkt", "wantlistSize": 4094, "overflowSize": 2}
2025-02-05T07:30:49.498Z        ERROR   core    core/listener.go:171    listener: resubscribe   {"err": "already subscribed to new blocks"}
2025-02-05T07:30:54.498Z        ERROR   core    core/listener.go:171    listener: resubscribe   {"err": "already subscribed to new blocks"}
2025-02-05T07:30:59.499Z        ERROR   core    core/listener.go:171    listener: resubscribe   {"err": "already subscribed to new blocks"}
```

### Is the node "stuck"? Has it stopped syncing?

Yes

### Notes

We encountered an issue today on our testnet bridge node. After restarting the archive RPC, our bridge node got stuck and could not recover on its own. However, a manual service restart resolved the issue.

Pull request description:
fix(core/fetcher): resubscribe if consensus node goes offline
FIxes #4090. 



Patch:
diff --git a/core/fetcher.go b/core/fetcher.go
index 6f049cc7..e47b913d 100644
--- a/core/fetcher.go
+++ b/core/fetcher.go
@@ -13,6 +13,7 @@ import (
 	coregrpc "github.com/tendermint/tendermint/rpc/grpc"
 	"github.com/tendermint/tendermint/types"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
 
 	libhead "github.com/celestiaorg/go-header"
 )
@@ -173,9 +174,12 @@ func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan types
 
 	subscription, err := f.client.SubscribeNewHeights(ctx, &coregrpc.SubscribeNewHeightsRequest{})
 	if err != nil {
+		close(f.doneCh)
+		f.isListeningForBlocks.Store(false)
 		return nil, err
 	}
 
+	log.Debug("created a subscription. Start listening for new blocks...")
 	signedBlockCh := make(chan types.EventDataSignedBlock)
 	go func() {
 		defer close(f.doneCh)
@@ -189,6 +193,12 @@ func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan types
 				resp, err := subscription.Recv()
 				if err != nil {
 					log.Errorw("fetcher: error receiving new height", "err", err.Error())
+					_, ok := status.FromError(err) // parsing the gRPC error
+					if ok {
+						// ok means that err contains a gRPC status error.
+						// move on another round of resubscribing.
+						return
+					}
 					continue
 				}
 				withTimeout, ctxCancel := context.WithTimeout(ctx, 10*time.Second)


Test patch:
diff --git a/core/fetcher_test.go b/core/fetcher_test.go
index 8d765949..0f2c4dc9 100644
--- a/core/fetcher_test.go
+++ b/core/fetcher_test.go
@@ -2,12 +2,15 @@ package core
 
 import (
 	"context"
+	"errors"
 	"net"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	coregrpc "github.com/tendermint/tendermint/rpc/grpc"
+	"google.golang.org/grpc"
 )
 
 func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
@@ -40,3 +43,130 @@ func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
 	}
 	require.NoError(t, fetcher.Stop(ctx))
 }
+
+type mockAPIService struct {
+	coregrpc.UnimplementedBlockAPIServer
+
+	grpcServer *grpc.Server
+	fetcher    *BlockFetcher
+}
+
+func (m *mockAPIService) SubscribeNewHeights(
+	_ *coregrpc.SubscribeNewHeightsRequest,
+	srv coregrpc.BlockAPI_SubscribeNewHeightsServer,
+) error {
+	for i := 0; i < 20; i++ {
+		b, err := m.fetcher.GetBlock(context.Background(), int64(i))
+		if err != nil {
+			return err
+		}
+		err = srv.Send(&coregrpc.NewHeightEvent{Height: b.Header.Height, Hash: b.Header.Hash()})
+		if err != nil {
+			return err
+		}
+		time.Sleep(time.Second)
+	}
+	return nil
+}
+
+func (m *mockAPIService) BlockByHeight(
+	req *coregrpc.BlockByHeightRequest,
+	srv coregrpc.BlockAPI_BlockByHeightServer,
+) error {
+	b, err := m.fetcher.client.BlockByHeight(context.Background(), &coregrpc.BlockByHeightRequest{Height: req.Height})
+	if err != nil {
+		return err
+	}
+	data, err := b.Recv()
+	if err != nil {
+		return err
+	}
+	err = srv.Send(data)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *mockAPIService) Start() error {
+	listener, err := net.Listen("tcp", ":50051")
+	if err != nil {
+		return err
+	}
+
+	grpcServer := grpc.NewServer()
+	m.grpcServer = grpcServer
+	coregrpc.RegisterBlockAPIServer(grpcServer, m)
+	go func() {
+		err = grpcServer.Serve(listener)
+		if err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			panic(err)
+		}
+	}()
+	return nil
+}
+
+func (m *mockAPIService) Stop() error {
+	m.grpcServer.Stop()
+	return nil
+}
+
+func (m *mockAPIService) generateBlocksWithHeights(ctx context.Context, t *testing.T) {
+	cfg := DefaultTestConfig()
+	fetcher, cctx := createCoreFetcher(t, cfg)
+	m.fetcher = fetcher
+	generateNonEmptyBlocks(t, ctx, fetcher, cfg, cctx)
+	require.NoError(t, fetcher.Stop(ctx))
+}
+
+// TestStart_SubscribeNewBlockEvent_Resubscription ensures that subscription will not stuck in case
+// gRPC server was stopped.
+func TestStart_SubscribeNewBlockEvent_Resubscription(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+	t.Cleanup(cancel)
+	m := &mockAPIService{}
+	m.generateBlocksWithHeights(ctx, t)
+
+	require.NoError(t, m.Start())
+
+	client := newTestClient(t, "localhost", "50051")
+
+	fetcher, err := NewBlockFetcher(client)
+	require.NoError(t, err)
+	// subscribe to block event to get blocks
+	newBlockChan, err := fetcher.SubscribeNewBlockEvent(ctx)
+	require.NoError(t, err)
+
+	select {
+	case newBlockFromChan := <-newBlockChan:
+		h := newBlockFromChan.Header.Height
+		_, err := fetcher.GetSignedBlock(ctx, h)
+		require.NoError(t, err)
+	case <-ctx.Done():
+		require.NoError(t, ctx.Err())
+	}
+
+	require.NoError(t, m.Stop())
+
+	// stopping the server sends an error with the status code
+	// to client, so the subscription loop will be finished.
+	// check that newBlockChan was closed
+	_, ok := <-newBlockChan
+	require.False(t, ok)
+
+	// start server and try to get a new subscription
+	require.NoError(t, m.Start())
+	newBlockChan, err = fetcher.SubscribeNewBlockEvent(ctx)
+	require.NoError(t, err)
+	select {
+	case newBlockFromChan := <-newBlockChan:
+		h := newBlockFromChan.Header.Height
+		_, err := fetcher.GetSignedBlock(ctx, h)
+		require.NoError(t, err)
+	case <-ctx.Done():
+		require.NoError(t, ctx.Err())
+	}
+	require.NoError(t, m.Stop())
+	require.NoError(t, m.fetcher.Stop(ctx))
+	require.NoError(t, fetcher.Stop(ctx))
+}