chain notify return error (#4677)

Makefile

@@ -11,7 +11,7 @@ deps:
 lint:
 	go run ./build/*.go lint
 
-test:
+test: test-venus-shared
 	go run ./build/*.go test -timeout=30m
 
 # WARNING THIS BUILDS A GO PLUGIN AND PLUGINS *DO NOT* WORK ON WINDOWS SYSTEMS
@@ -64,8 +64,6 @@ inline-gen:
 test-venus-shared:
 	cd venus-shared && go test -covermode=set ./...
 
-test: gogen test-venus-shared
-
 compatible-all: compatible-api compatible-actor
 
 compatible-api: api-checksum api-diff

pkg/events/cache.go

@@ -12,7 +12,7 @@ import (
 )
 
 type uncachedAPI interface {
-	ChainNotify(context.Context) <-chan []*types.HeadChange
+	ChainNotify(context.Context) (<-chan []*types.HeadChange, error)
 	ChainGetPath(ctx context.Context, from, to types.TipSetKey) ([]*types.HeadChange, error)
 	StateSearchMsg(ctx context.Context, from types.TipSetKey, msg cid.Cid, limit abi.ChainEpoch, allowReplaced bool) (*types.MsgLookup, error)
 

pkg/events/eventAPI.go

@@ -17,7 +17,7 @@ type TipSetObserver interface {
 }
 
 type IEvent interface {
-	ChainNotify(context.Context) <-chan []*types.HeadChange
+	ChainNotify(context.Context) (<-chan []*types.HeadChange, error)
 	ChainGetBlockMessages(context.Context, cid.Cid) (*types.BlockMessages, error)
 	ChainGetTipSetByHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)
 	ChainGetTipSetAfterHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)

pkg/events/events_test.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"time"
 
+	"golang.org/x/xerrors"
+
 	"sync"
 	"testing"
 
@@ -70,7 +72,7 @@ func newFakeCS(t *testing.T) *fakeCS {
 		cancel:     cancel,
 	}
 	require.NoError(t, fcs.tsc.add(fcs.makeTs(t, nil, 1, dummyCid)))
-	fcs.loopNotify(ctx)
+	require.NoError(t, fcs.loopNotify(ctx))
 	return fcs
 }
 
@@ -83,8 +85,11 @@ func (fcs *fakeCS) stop() {
 
 // our observe use a timer and call 'chainhead' to observe chain head change
 // to 'PASS' these tests, we must call 'ChainNotify' to start 'waitSub'
-func (fcs *fakeCS) loopNotify(ctx context.Context) {
-	head := fcs.ChainNotify(ctx)
+func (fcs *fakeCS) loopNotify(ctx context.Context) error {
+	head, err := fcs.ChainNotify(ctx)
+	if err != nil {
+		return err
+	}
 	go func() {
 		for {
 			select {
@@ -94,6 +99,8 @@ func (fcs *fakeCS) loopNotify(ctx context.Context) {
 			}
 		}
 	}()
+
+	return nil
 }
 
 func (fcs *fakeCS) ChainHead(ctx context.Context) (*types.TipSet, error) {
@@ -218,16 +225,15 @@ func (fcs *fakeCS) makeTs(t *testing.T, parents []cid.Cid, h abi.ChainEpoch, msg
 	return ts
 }
 
-func (fcs *fakeCS) ChainNotify(ctx context.Context) <-chan []*types.HeadChange {
+func (fcs *fakeCS) ChainNotify(ctx context.Context) (<-chan []*types.HeadChange, error) {
 	fcs.mu.Lock()
 	defer fcs.mu.Unlock()
 	fcs.callNumber["ChainNotify"] = fcs.callNumber["ChainNotify"] + 1
 
 	out := make(chan []*types.HeadChange, 1)
 	if fcs.subCh != nil {
 		close(out)
-		fcs.t.Error("already subscribed to notifications")
-		return out
+		return out, xerrors.Errorf("already subscribed to notifications")
 	}
 
 	best, err := fcs.tsc.ChainHead(ctx)
@@ -239,7 +245,7 @@ func (fcs *fakeCS) ChainNotify(ctx context.Context) <-chan []*types.HeadChange {
 	fcs.subCh = out
 	close(fcs.waitSub)
 
-	return out
+	return out, nil
 }
 
 func (fcs *fakeCS) ChainGetBlockMessages(ctx context.Context, blk cid.Cid) (*types.BlockMessages, error) {