service/header/core_listener: Fix bug where old block headers are spammed to headersub by bridge nodes

core/fetcher.go

@@ -177,3 +177,14 @@ func (f *BlockFetcher) UnsubscribeNewBlockEvent(ctx context.Context) error {
 
 	return f.client.Unsubscribe(ctx, newBlockSubscriber, newBlockEventQuery)
 }
+
+// IsSyncing returns the sync status of the Core connection: true for
+// syncing, and false for already caught up. It can also return an error
+// in the case of a failed status request.
+func (f *BlockFetcher) IsSyncing(ctx context.Context) (bool, error) {
+	resp, err := f.client.Status(ctx)
+	if err != nil {
+		return false, err
+	}
+	return resp.SyncInfo.CatchingUp, nil
+}

service/header/core_listener.go

@@ -46,8 +46,9 @@ func (cl *CoreListener) Start(ctx context.Context) error {
 		return err
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	go cl.listen(ctx, sub)
+	clCtx, cancel := context.WithCancel(context.Background())
+
+	go cl.listen(clCtx, sub)
 	cl.cancel = cancel
 	return nil
 }
@@ -56,14 +57,20 @@ func (cl *CoreListener) Start(ctx context.Context) error {
 func (cl *CoreListener) Stop(ctx context.Context) error {
 	cl.cancel()
 	cl.cancel = nil
-	return cl.fetcher.UnsubscribeNewBlockEvent(ctx)
+
+	err := cl.fetcher.UnsubscribeNewBlockEvent(ctx)
+	if err != nil {
+		log.Errorw("core-listener: failed to unsubscribe from new block events", "err", err)
+	}
+	return nil
 }
 
 // listen kicks off a loop, listening for new block events from Core,
 // generating ExtendedHeaders and broadcasting them to the header-sub
 // gossipsub network.
 func (cl *CoreListener) listen(ctx context.Context, sub <-chan *types.Block) {
 	defer log.Info("core-listener: listening stopped")
+
 	for {
 		select {
 		case b, ok := <-sub:
@@ -83,15 +90,24 @@ func (cl *CoreListener) listen(ctx context.Context, sub <-chan *types.Block) {
 				return
 			}
 
-			// broadcast new ExtendedHeader
-			err = cl.bcast.Broadcast(ctx, eh)
+			// only broadcast new ExtendedHeaders if core has finished syncing
+			// in order to prevent spamming of old block headers to `headersub`
+			syncing, err := cl.fetcher.IsSyncing(ctx)
 			if err != nil {
-				var pserr pubsub.ValidationError
-				// TODO(@Wondertan): Log ValidationIgnore cases as well, once headr duplication issue is fixed
-				if errors.As(err, &pserr) && pserr.Reason != pubsub.RejectValidationIgnored {
-					log.Errorw("core-listener: broadcasting next header", "height", eh.Height,
-						"err", err)
-					return
+				log.Errorw("core-listener: fetching sync status from core", "err", err)
+				return
+			}
+			if !syncing {
+				// broadcast new ExtendedHeader
+				err = cl.bcast.Broadcast(ctx, eh)
+				if err != nil {
+					var pserr pubsub.ValidationError
+					// TODO(@Wondertan): Log ValidationIgnore cases as well, once headr duplication issue is fixed
+					if errors.As(err, &pserr) && pserr.Reason != pubsub.RejectValidationIgnored {
+						log.Errorw("core-listener: broadcasting next header", "height", eh.Height,
+							"err", err)
+						return
+					}
 				}
 			}
 		case <-ctx.Done():