fix: broken shutdown for services on fraud proofs receival (celestiao…

…rg#1220) Closes celestiaorg#1205 Co-authored-by: Wondertan <[email protected]>
vgonkivs · Oct 17, 2022 · d239616 · d239616
1 parent 7fcb0f3
commit d239616
Show file tree

Hide file tree

Showing 9 changed files with 23 additions and 29 deletions.
diff --git a/nodebuilder/daser/module.go b/nodebuilder/daser/module.go
@@ -7,7 +7,6 @@ import (
 
 	"github.com/celestiaorg/celestia-node/das"
 	"github.com/celestiaorg/celestia-node/fraud"
-	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	fraudServ "github.com/celestiaorg/celestia-node/nodebuilder/fraud"
 	"github.com/celestiaorg/celestia-node/nodebuilder/node"
 )
@@ -19,9 +18,8 @@ func ConstructModule(tp node.Type) fx.Option {
 			"daser",
 			fx.Provide(fx.Annotate(
 				NewDASer,
-				fx.OnStart(func(ctx context.Context, lc fx.Lifecycle, fservice fraudServ.Module, das *das.DASer) error {
-					lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
-					return fraudServ.Lifecycle(ctx, lifecycleCtx, fraud.BadEncoding, fservice,
+				fx.OnStart(func(startCtx, ctx context.Context, fservice fraudServ.Module, das *das.DASer) error {
+					return fraudServ.Lifecycle(startCtx, ctx, fraud.BadEncoding, fservice,
 						das.Start, das.Stop)
 				}),
 				fx.OnStop(func(ctx context.Context, das *das.DASer) error {

diff --git a/nodebuilder/header/module.go b/nodebuilder/header/module.go
@@ -11,7 +11,6 @@ import (
 	"github.com/celestiaorg/celestia-node/header/p2p"
 	"github.com/celestiaorg/celestia-node/header/store"
 	"github.com/celestiaorg/celestia-node/header/sync"
-	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	fraudServ "github.com/celestiaorg/celestia-node/nodebuilder/fraud"
 	"github.com/celestiaorg/celestia-node/nodebuilder/node"
 	"github.com/celestiaorg/celestia-node/params"
@@ -43,7 +42,7 @@ func ConstructModule(tp node.Type, cfg *Config) fx.Option {
 		}),
 		fx.Provide(fx.Annotate(
 			newSyncer,
-			fx.OnStart(func(ctx context.Context, lc fx.Lifecycle, fservice fraudServ.Module, syncer *sync.Syncer) error {
+			fx.OnStart(func(startCtx, ctx context.Context, fservice fraudServ.Module, syncer *sync.Syncer) error {
 				syncerStartFunc := func(ctx context.Context) error {
 					err := syncer.Start(ctx)
 					switch err {
@@ -55,8 +54,7 @@ func ConstructModule(tp node.Type, cfg *Config) fx.Option {
 					}
 					return nil
 				}
-				lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
-				return fraudServ.Lifecycle(ctx, lifecycleCtx, fraud.BadEncoding, fservice,
+				return fraudServ.Lifecycle(startCtx, ctx, fraud.BadEncoding, fservice,
 					syncerStartFunc, syncer.Stop)
 			}),
 			fx.OnStop(func(ctx context.Context, syncer *sync.Syncer) error {

diff --git a/nodebuilder/module.go b/nodebuilder/module.go
@@ -5,6 +5,7 @@ import (
 
 	"go.uber.org/fx"
 
+	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	"github.com/celestiaorg/celestia-node/nodebuilder/core"
 	"github.com/celestiaorg/celestia-node/nodebuilder/daser"
 	"github.com/celestiaorg/celestia-node/nodebuilder/fraud"
@@ -21,7 +22,9 @@ func ConstructModule(tp node.Type, cfg *Config, store Store) fx.Option {
 	baseComponents := fx.Options(
 		fx.Provide(params.DefaultNetwork),
 		fx.Provide(params.BootstrappersFor),
-		fx.Provide(context.Background),
+		fx.Provide(func(lc fx.Lifecycle) context.Context {
+			return fxutil.WithLifecycle(context.Background(), lc)
+		}),
 		fx.Supply(cfg),
 		fx.Supply(store.Config),
 		fx.Provide(store.Datastore),

diff --git a/nodebuilder/p2p/bitswap.go b/nodebuilder/p2p/bitswap.go
@@ -14,7 +14,6 @@ import (
 	routinghelpers "github.com/libp2p/go-libp2p-routing-helpers"
 	"go.uber.org/fx"
 
-	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	nparams "github.com/celestiaorg/celestia-node/params"
 )
 
@@ -29,9 +28,8 @@ const (
 
 // DataExchange provides a constructor for IPFS block's DataExchange over BitSwap.
 func DataExchange(params bitSwapParams) (exchange.Interface, blockstore.Blockstore, error) {
-	ctx := fxutil.WithLifecycle(params.Ctx, params.Lc)
 	bs, err := blockstore.CachedBlockstore(
-		ctx,
+		params.Ctx,
 		blockstore.NewBlockstore(params.Ds),
 		blockstore.CacheOpts{
 			HasBloomFilterSize:   defaultBloomFilterSize,
@@ -44,7 +42,7 @@ func DataExchange(params bitSwapParams) (exchange.Interface, blockstore.Blocksto
 	}
 	prefix := protocol.ID(fmt.Sprintf("/celestia/%s", params.Net))
 	return bitswap.New(
-		ctx,
+		params.Ctx,
 		network.NewFromIpfsHost(params.Host, &routinghelpers.Null{}, network.Prefix(prefix)),
 		bs,
 		bitswap.ProvideEnabled(false),
@@ -60,7 +58,6 @@ type bitSwapParams struct {
 
 	Ctx  context.Context
 	Net  nparams.Network
-	Lc   fx.Lifecycle
 	Host host.Host
 	Ds   datastore.Batching
 }
diff --git a/nodebuilder/p2p/pubsub.go b/nodebuilder/p2p/pubsub.go
@@ -8,8 +8,6 @@ import (
 	pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
 	"go.uber.org/fx"
 	"golang.org/x/crypto/blake2b"
-
-	"github.com/celestiaorg/celestia-node/libs/fxutil"
 )
 
 // PubSub provides a constructor for PubSub protocol with GossipSub routing.
@@ -34,7 +32,7 @@ func PubSub(cfg Config, params pubSubParams) (*pubsub.PubSub, error) {
 	}
 
 	return pubsub.NewGossipSub(
-		fxutil.WithLifecycle(params.Ctx, params.Lc),
+		params.Ctx,
 		params.Host,
 		opts...,
 	)
@@ -49,6 +47,5 @@ type pubSubParams struct {
 	fx.In
 
 	Ctx  context.Context
-	Lc   fx.Lifecycle
 	Host host.Host
 }
diff --git a/nodebuilder/p2p/routing.go b/nodebuilder/p2p/routing.go
@@ -11,7 +11,6 @@ import (
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	"go.uber.org/fx"
 
-	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	nparams "github.com/celestiaorg/celestia-node/params"
 )
 
@@ -42,7 +41,7 @@ func PeerRouting(cfg Config, params routingParams) (routing.PeerRouting, error)
 		)
 	}
 
-	d, err := dht.New(fxutil.WithLifecycle(params.Ctx, params.Lc), params.Host, opts...)
+	d, err := dht.New(params.Ctx, params.Host, opts...)
 	if err != nil {
 		return nil, err
 	}

diff --git a/nodebuilder/state/module.go b/nodebuilder/state/module.go
@@ -7,7 +7,6 @@ import (
 	"go.uber.org/fx"
 
 	"github.com/celestiaorg/celestia-node/fraud"
-	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	fraudServ "github.com/celestiaorg/celestia-node/nodebuilder/fraud"
 	"github.com/celestiaorg/celestia-node/nodebuilder/node"
 	"github.com/celestiaorg/celestia-node/state"
@@ -27,9 +26,8 @@ func ConstructModule(tp node.Type, cfg *Config) fx.Option {
 		fx.Provide(Keyring),
 		fx.Provide(fx.Annotate(
 			CoreAccessor,
-			fx.OnStart(func(ctx context.Context, lc fx.Lifecycle, fservice fraudServ.Module, ca *state.CoreAccessor) error {
-				lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
-				return fraudServ.Lifecycle(ctx, lifecycleCtx, fraud.BadEncoding, fservice, ca.Start, ca.Stop)
+			fx.OnStart(func(startCtx, ctx context.Context, fservice fraudServ.Module, ca *state.CoreAccessor) error {
+				return fraudServ.Lifecycle(startCtx, ctx, fraud.BadEncoding, fservice, ca.Start, ca.Stop)
 			}),
 			fx.OnStop(func(ctx context.Context, ca *state.CoreAccessor) error {
 				return ca.Stop(ctx)

diff --git a/nodebuilder/tests/fraud_test.go b/nodebuilder/tests/fraud_test.go
@@ -107,7 +107,7 @@ func TestFraudProofSyncing(t *testing.T) {
 	store := nodebuilder.MockStore(t, cfg)
 	bridge := sw.NewNodeWithStore(node.Bridge, store, core.WithHeaderConstructFn(header.FraudMaker(t, 10)))
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
+	ctx, cancel := context.WithTimeout(context.Background(), swamp.DefaultTestTimeout)
 	t.Cleanup(cancel)
 
 	err := bridge.Start(ctx)
@@ -123,21 +123,24 @@ func TestFraudProofSyncing(t *testing.T) {
 	lightCfg := nodebuilder.DefaultConfig(node.Light)
 	lightCfg.P2P.RoutingTableRefreshPeriod = defaultTimeInterval
 	lightCfg.Share.DiscoveryInterval = defaultTimeInterval
-	ln := sw.NewNodeWithStore(node.Light, nodebuilder.MockStore(t, lightCfg),
-		nodebuilder.WithBootstrappers([]peer.AddrInfo{*addr}))
+	lightCfg.Header.TrustedPeers = append(lightCfg.Header.TrustedPeers, addrs[0].String())
+	ln := sw.NewNodeWithStore(node.Light, nodebuilder.MockStore(t, lightCfg))
 
 	require.NoError(t, full.Start(ctx))
+	require.NoError(t, ln.Start(ctx))
 	subsFn, err := full.FraudServ.Subscribe(fraud.BadEncoding)
 	require.NoError(t, err)
 	defer subsFn.Cancel()
 	_, err = subsFn.Proof(ctx)
 	require.NoError(t, err)
 
-	require.NoError(t, ln.Start(ctx))
 	// internal subscription for the fraud proof is done in order to ensure that light node
 	// receives the BEFP.
 	subsLn, err := ln.FraudServ.Subscribe(fraud.BadEncoding)
 	require.NoError(t, err)
+
+	err = ln.Host.Connect(ctx, *host.InfoFromHost(full.Host))
+	require.NoError(t, err)
 	_, err = subsLn.Proof(ctx)
 	require.NoError(t, err)
 	subsLn.Cancel()

diff --git a/state/core_access.go b/state/core_access.go
@@ -98,7 +98,8 @@ func (ca *CoreAccessor) Start(ctx context.Context) error {
 func (ca *CoreAccessor) Stop(context.Context) error {
 	defer ca.cancel()
 	if ca.coreConn == nil {
-		return fmt.Errorf("core-access: no connection found to close")
+		log.Warn("no connection found to close")
+		return nil
 	}
 	// close out core connection
 	err := ca.coreConn.Close()