Merge pull request #612 from shutter-network/fix/graceful-close

close p2p host, cancel subscriptions for closing service gracefully when context is cancelled

Author: blockchainluffy

rolling-shutter/keyper/eonpkhandler.go

@@ -66,6 +66,7 @@ func (pkh *eonPubKeyHandler) loop(ctx context.Context) error {
 		}
 		select {
 		case <-ctx.Done():
+			t.Stop()
 			return ctx.Err()
 		case <-t.C:
 		}

rolling-shutter/medley/chainsync/syncer/eonpubkey.go

@@ -60,9 +60,13 @@ func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) erro
 	if err != nil {
 		return err
 	}
-	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchNewEonPubkey(ctx, subs.Err())
+		err := s.watchNewEonPubkey(ctx, subs.Err())
+		if err != nil {
+			s.Log.Error("error watching new eon pubkey", err.Error())
+		}
+		subs.Unsubscribe()
+		return err
 	})
 	return nil
 }

rolling-shutter/medley/chainsync/syncer/keyperset.go

@@ -70,9 +70,13 @@ func (s *KeyperSetSyncer) Start(ctx context.Context, runner service.Runner) erro
 	if err != nil {
 		return err
 	}
-	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchNewKeypersService(ctx, subs.Err())
+		err := s.watchNewKeypersService(ctx, subs.Err())
+		if err != nil {
+			s.Log.Error("error watching new keypers", err.Error())
+		}
+		subs.Unsubscribe()
+		return err
 	})
 	return nil
 }

rolling-shutter/medley/chainsync/syncer/shutterstate.go

@@ -53,17 +53,21 @@ func (s *ShutterStateSyncer) Start(ctx context.Context, runner service.Runner) e
 	if err != nil {
 		return err
 	}
-	runner.Defer(subs.Unsubscribe)
 
 	s.unpausedCh = make(chan *bindings.KeyperSetManagerUnpaused)
 	subsUnpaused, err := s.Contract.WatchUnpaused(watchOpts, s.unpausedCh)
 	if err != nil {
 		return err
 	}
-	runner.Defer(subsUnpaused.Unsubscribe)
 
 	runner.Go(func() error {
-		return s.watchPaused(ctx, subs.Err(), subsUnpaused.Err())
+		err := s.watchPaused(ctx, subs.Err(), subsUnpaused.Err())
+		if err != nil {
+			s.Log.Error("error watching paused", err.Error())
+		}
+		subs.Unsubscribe()
+		subsUnpaused.Unsubscribe()
+		return err
 	})
 	return nil
 }

rolling-shutter/medley/chainsync/syncer/unsafehead.go

@@ -31,9 +31,13 @@ func (s *UnsafeHeadSyncer) Start(ctx context.Context, runner service.Runner) err
 	if err != nil {
 		return err
 	}
-	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchLatestUnsafeHead(ctx, subs.Err())
+		err := s.watchLatestUnsafeHead(ctx, subs.Err())
+		if err != nil {
+			s.Log.Error("error watching latest unsafe head", err.Error())
+		}
+		subs.Unsubscribe()
+		return err
 	})
 	return nil
 }

rolling-shutter/p2p/p2p.go

@@ -111,6 +111,19 @@ func (p *P2PNode) Run(
 		return err
 	}
 
+	runner.Go(func() error {
+		<-ctx.Done()
+		log.Debug().Msg("stopping host when context is done")
+		if err := p.host.Close(); err != nil {
+			log.Error().Err(err).Msg("error closing host")
+		}
+		if err := p.dht.Close(); err != nil {
+			log.Error().Err(err).Msg("error closing dht")
+		}
+		log.Debug().Msg("host closed")
+		return nil
+	})
+
 	for topicName := range topicValidators {
 		validator := topicValidators.GetCombinedValidator(topicName)
 		if err := p.pubSub.RegisterTopicValidator(topicName, validator); err != nil {
@@ -158,8 +171,7 @@ func (p *P2PNode) Run(
 	runner.Go(func() error {
 		log.Info().Str("namespace", p.config.DiscoveryNamespace).Msg("starting advertizing discovery node")
 		util.Advertise(ctx, p.discovery, p.config.DiscoveryNamespace)
-		<-ctx.Done()
-		return ctx.Err()
+		return nil
 	})
 	runner.Go(func() error {
 		return findPeers(ctx, p.host, p.discovery, p.config.DiscoveryNamespace)

rolling-shutter/p2p/topic.go

@@ -5,6 +5,7 @@ import (
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/rs/zerolog/log"
 )
 
 // gossipRoom represents a subscription to a single PubSub topic. Messages
@@ -42,6 +43,11 @@ func (room *gossipRoom) readLoop(ctx context.Context, messages chan *pubsub.Mess
 		select {
 		case messages <- msg:
 		case <-ctx.Done():
+			log.Debug().Msg("subscription canceled, closing read loop")
+			room.subscription.Cancel()
+			if err := room.topic.Close(); err != nil {
+				return err
+			}
 			return ctx.Err()
 		}
 	}