Merge pull request #635 from shutter-network/fix/client-close

fix: keyper hanging issue

Author: blockchainluffy

rolling-shutter/keyperimpl/gnosis/keyper.go

@@ -295,17 +295,29 @@ func (kpr *Keyper) processInputs(ctx context.Context) error {
 	}
 }
 
-func (kpr *Keyper) channelNewBlock(_ context.Context, ev *syncevent.LatestBlock) error {
-	kpr.newBlocks <- ev
-	return nil
+func (kpr *Keyper) channelNewBlock(ctx context.Context, ev *syncevent.LatestBlock) error {
+	select {
+	case kpr.newBlocks <- ev:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
-func (kpr *Keyper) channelNewKeyperSet(_ context.Context, ev *syncevent.KeyperSet) error {
-	kpr.newKeyperSets <- ev
-	return nil
+func (kpr *Keyper) channelNewKeyperSet(ctx context.Context, ev *syncevent.KeyperSet) error {
+	select {
+	case kpr.newKeyperSets <- ev:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
-func (kpr *Keyper) channelNewEonPublicKey(_ context.Context, key keyper.EonPublicKey) error {
-	kpr.newEonPublicKeys <- key
-	return nil
+func (kpr *Keyper) channelNewEonPublicKey(ctx context.Context, key keyper.EonPublicKey) error {
+	select {
+	case kpr.newEonPublicKeys <- key:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }

rolling-shutter/keyperimpl/gnosis/newslot.go

@@ -266,7 +266,12 @@ func (kpr *Keyper) triggerDecryption(
 		Int("num-identities", len(trigger.IdentityPreimages)).
 		Int64("tx-pointer", txPointer).
 		Msg("sending decryption trigger")
-	kpr.decryptionTriggerChannel <- event
+
+	select {
+	case kpr.decryptionTriggerChannel <- event:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 
 	return nil
 }

rolling-shutter/keyperimpl/shutterservice/keyper.go

@@ -274,17 +274,29 @@ func (kpr *Keyper) processInputs(ctx context.Context) error {
 	}
 }
 
-func (kpr *Keyper) channelNewEonPublicKey(_ context.Context, key keyper.EonPublicKey) error {
-	kpr.newEonPublicKeys <- key
-	return nil
+func (kpr *Keyper) channelNewEonPublicKey(ctx context.Context, key keyper.EonPublicKey) error {
+	select {
+	case kpr.newEonPublicKeys <- key:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
-func (kpr *Keyper) channelNewBlock(_ context.Context, ev *syncevent.LatestBlock) error {
-	kpr.newBlocks <- ev
-	return nil
+func (kpr *Keyper) channelNewBlock(ctx context.Context, ev *syncevent.LatestBlock) error {
+	select {
+	case kpr.newBlocks <- ev:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
-func (kpr *Keyper) channelNewKeyperSet(_ context.Context, ev *syncevent.KeyperSet) error {
-	kpr.newKeyperSets <- ev
-	return nil
+func (kpr *Keyper) channelNewKeyperSet(ctx context.Context, ev *syncevent.KeyperSet) error {
+	select {
+	case kpr.newKeyperSets <- ev:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }

rolling-shutter/medley/chainsync/client.go

@@ -146,7 +146,13 @@ func (s *Client) ChainID(ctx context.Context) (*big.Int, error) {
 	return s.chainID, nil
 }
 
-func (s *Client) Start(_ context.Context, runner service.Runner) error {
+func (s *Client) Start(ctx context.Context, runner service.Runner) error {
+	runner.Go(func() error {
+		<-ctx.Done()
+		s.Client.Close()
+		s.log.Debug("chainsync client closed")
+		return nil
+	})
 	return runner.StartService(s.getServices()...)
 }
 

rolling-shutter/medley/channel/channel.go

@@ -12,7 +12,11 @@ func Forward[T any](ctx context.Context,
 			if !ok {
 				return nil
 			}
-			send <- val
+			select {
+			case send <- val:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
 		case <-ctx.Done():
 			return ctx.Err()
 		}

rolling-shutter/medley/slotticker/slotticker.go

@@ -92,7 +92,11 @@ func (t *SlotTicker) run(ctx context.Context) error {
 
 		timeToNextSlot := nextTickTime.Sub(now)
 		timer.Reset(timeToNextSlot)
-		<-timer.C
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
 
 		if err := t.tick(ctx, nextSlotNumber); err != nil {
 			return err

rolling-shutter/p2p/bootstrap.go

@@ -100,6 +100,7 @@ func bootstrap(
 		_, err := retry.FunctionCall(
 			ctx,
 			f,
+			retry.NumberOfRetries(10),
 			retry.StopOnErrors(errInsufficientBootstrpConfigured),
 			retry.Interval(2*time.Second))
 		if err != nil {

rolling-shutter/p2p/p2p.go

@@ -103,10 +103,6 @@ func (p *P2PNode) Run(
 	p.mux.Lock()
 	defer p.mux.Unlock()
 
-	runner.Defer(func() {
-		close(p.GossipMessages)
-	})
-
 	if err := p.init(ctx); err != nil {
 		return err
 	}
@@ -120,6 +116,7 @@ func (p *P2PNode) Run(
 		if err := p.dht.Close(); err != nil {
 			log.Error().Err(err).Msg("error closing dht")
 		}
+		close(p.GossipMessages)
 		log.Debug().Msg("host closed")
 		return nil
 	})