add logging when event subscription fails

Author: blockchainluffy

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

@@ -57,13 +57,12 @@ func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) erro
 	}
 	s.keyBroadcastCh = make(chan *bindings.KeyBroadcastContractEonKeyBroadcast, channelSize)
 	subs, err := s.KeyBroadcast.WatchEonKeyBroadcast(watchOpts, s.keyBroadcastCh)
-	// FIXME: what to do on subs.Error()
 	if err != nil {
 		return err
 	}
 	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchNewEonPubkey(ctx)
+		return s.watchNewEonPubkey(ctx, subs.Err())
 	})
 	return nil
 }
@@ -127,7 +126,7 @@ func (s *EonPubKeySyncer) GetEonPubKeyForEon(ctx context.Context, opts *bind.Cal
 	}, nil
 }
 
-func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
+func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newEonKey, ok := <-s.keyBroadcastCh:
@@ -148,6 +147,8 @@ func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
 					err.Error(),
 				)
 			}
+		case err := <-subsErr:
+			s.Log.Error("subscription error for watchNewEonPubkey", err.Error())
 		case <-ctx.Done():
 			return ctx.Err()
 		}

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

@@ -67,13 +67,12 @@ func (s *KeyperSetSyncer) Start(ctx context.Context, runner service.Runner) erro
 	}
 	s.keyperAddedCh = make(chan *bindings.KeyperSetManagerKeyperSetAdded, channelSize)
 	subs, err := s.Contract.WatchKeyperSetAdded(watchOpts, s.keyperAddedCh)
-	// FIXME: what to do on subs.Error()
 	if err != nil {
 		return err
 	}
 	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchNewKeypersService(ctx)
+		return s.watchNewKeypersService(ctx, subs.Err())
 	})
 	return nil
 }
@@ -205,7 +204,7 @@ func (s *KeyperSetSyncer) newEvent(
 	}, nil
 }
 
-func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context) error {
+func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newKeypers, ok := <-s.keyperAddedCh:
@@ -235,6 +234,8 @@ func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context) error {
 					err.Error(),
 				)
 			}
+		case err := <-subsErr:
+			s.Log.Error("subscription error for watchNewKeypersService", err.Error())
 		case <-ctx.Done():
 			return ctx.Err()
 		}

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

@@ -50,22 +50,20 @@ func (s *ShutterStateSyncer) Start(ctx context.Context, runner service.Runner) e
 	}
 	s.pausedCh = make(chan *bindings.KeyperSetManagerPaused)
 	subs, err := s.Contract.WatchPaused(watchOpts, s.pausedCh)
-	// FIXME: what to do on subs.Error()
 	if err != nil {
 		return err
 	}
 	runner.Defer(subs.Unsubscribe)
 
 	s.unpausedCh = make(chan *bindings.KeyperSetManagerUnpaused)
-	subs, err = s.Contract.WatchUnpaused(watchOpts, s.unpausedCh)
-	// FIXME: what to do on subs.Error()
+	subsUnpaused, err := s.Contract.WatchUnpaused(watchOpts, s.unpausedCh)
 	if err != nil {
 		return err
 	}
-	runner.Defer(subs.Unsubscribe)
+	runner.Defer(subsUnpaused.Unsubscribe)
 
 	runner.Go(func() error {
-		return s.watchPaused(ctx)
+		return s.watchPaused(ctx, subs.Err(), subsUnpaused.Err())
 	})
 	return nil
 }
@@ -89,7 +87,7 @@ func (s *ShutterStateSyncer) handle(ctx context.Context, ev *event.ShutterState)
 	}
 }
 
-func (s *ShutterStateSyncer) watchPaused(ctx context.Context) error {
+func (s *ShutterStateSyncer) watchPaused(ctx context.Context, subsErr <-chan error, subsErrUnpaused <-chan error) error {
 	isActive, err := s.pollIsActive(ctx)
 	if err != nil {
 		// XXX: this will fail everything, do we want that?
@@ -125,6 +123,10 @@ func (s *ShutterStateSyncer) watchPaused(ctx context.Context) error {
 			}
 			isActive = ev.Active
 			s.handle(ctx, ev)
+		case err := <-subsErr:
+			s.Log.Error("subscription error for watchPaused", err.Error())
+		case err := <-subsErrUnpaused:
+			s.Log.Error("subscription error for watchUnpaused", err.Error())
 		case <-ctx.Done():
 			return ctx.Err()
 		}

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

@@ -28,18 +28,17 @@ func (s *UnsafeHeadSyncer) Start(ctx context.Context, runner service.Runner) err
 	s.newLatestHeadCh = make(chan *types.Header, 1)
 
 	subs, err := s.Client.SubscribeNewHead(ctx, s.newLatestHeadCh)
-	// FIXME: what to do on subs.Error()
 	if err != nil {
 		return err
 	}
 	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchLatestUnsafeHead(ctx)
+		return s.watchLatestUnsafeHead(ctx, subs.Err())
 	})
 	return nil
 }
 
-func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context) error {
+func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newHeader, ok := <-s.newLatestHeadCh:
@@ -59,6 +58,8 @@ func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context) error {
 					err.Error(),
 				)
 			}
+		case err := <-subsErr:
+			s.Log.Error("subscription error for watchLatestUnsafeHead", err.Error())
 		case <-ctx.Done():
 			return ctx.Err()
 		}