pkg/beaconclient/capturehead.go

@@ -39,6 +39,8 @@ func (bc *BeaconClient) StopHeadTracking(ctx context.Context, skipSee bool) {
 			bc.HeadTracking.SseClient.Unsubscribe(bc.HeadTracking.MessagesCh)
 			bc.ReOrgTracking.SseClient.Unsubscribe(bc.ReOrgTracking.MessagesCh)
 			log.Info("Successfully unsubscribed to SSE client")
+			close(bc.ReOrgTracking.MessagesCh)
+			close(bc.HeadTracking.MessagesCh)
 		}
 		log.Info("Successfully stopped the head tracking service.")
 	default:

pkg/beaconclient/incomingsse.go

@@ -47,8 +47,7 @@ func handleIncomingSseEvent[P ProcessedEvents](ctx context.Context, eventHandler
 	for {
 		select {
 		case <-ctx.Done():
-			close(eventHandler.MessagesCh)
-			close(eventHandler.ErrorCh)
+			close(eventHandler.ProcessCh)
 			return
 		case message := <-eventHandler.MessagesCh:
 			// Message can be nil if its a keep-alive message

pkg/beaconclient/processevents.go

@@ -33,7 +33,6 @@ func (bc *BeaconClient) handleReorg(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			close(bc.ReOrgTracking.ProcessCh)
 			return
 		case reorg := <-bc.ReOrgTracking.ProcessCh:
 			log.WithFields(log.Fields{"reorg": reorg}).Debug("Received a new reorg message.")
@@ -55,7 +54,6 @@ func (bc *BeaconClient) handleHead(ctx context.Context, maxWorkers int) {
 	for {
 		select {
 		case <-ctx.Done():
-			close(bc.HeadTracking.ProcessCh)
 			close(workCh)
 			return
 		case head := <-bc.HeadTracking.ProcessCh: