replace receive slot with event stream

api/client/beacon/BUILD.bazel

@@ -6,12 +6,14 @@ go_library(
         "checkpoint.go",
         "client.go",
         "doc.go",
+        "health.go",
         "log.go",
     ],
     importpath = "github.com/prysmaticlabs/prysm/v5/api/client/beacon",
     visibility = ["//visibility:public"],
     deps = [
         "//api/client:go_default_library",
+        "//api/client/beacon/iface:go_default_library",
         "//api/server:go_default_library",
         "//api/server/structs:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
@@ -37,10 +39,12 @@ go_test(
     srcs = [
         "checkpoint_test.go",
         "client_test.go",
+        "health_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//api/client:go_default_library",
+        "//api/client/beacon/testing:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
@@ -54,5 +58,6 @@ go_test(
         "//testing/util:go_default_library",
         "//time/slots:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
     ],
 )

api/client/beacon/health.go

@@ -0,0 +1,55 @@
+package beacon
+
+import (
+	"context"
+	"sync"
+
+	"github.com/prysmaticlabs/prysm/v5/api/client/beacon/iface"
+)
+
+type NodeHealthTracker struct {
+	isHealthy  *bool
+	healthChan chan bool
+	node       iface.HealthNode
+	sync.RWMutex
+}
+
+func NewNodeHealthTracker(node iface.HealthNode) *NodeHealthTracker {
+	return &NodeHealthTracker{
+		node:       node,
+		healthChan: make(chan bool, 1),
+	}
+}
+
+// HealthUpdates provides a read-only channel for health updates.
+func (n *NodeHealthTracker) HealthUpdates() <-chan bool {
+	return n.healthChan
+}
+
+func (n *NodeHealthTracker) IsHealthy() bool {
+	n.RLock()
+	defer n.RUnlock()
+	if n.isHealthy == nil {
+		return false
+	}
+	return *n.isHealthy
+}
+
+func (n *NodeHealthTracker) CheckHealth(ctx context.Context) bool {
+	n.RLock()
+	newStatus := n.node.IsHealthy(ctx)
+	if n.isHealthy == nil {
+		n.isHealthy = &newStatus
+	}
+	isStatusChanged := newStatus != *n.isHealthy
+	n.RUnlock()
+
+	if isStatusChanged {
+		n.Lock()
+		// Double-check the condition to ensure it hasn't changed since the first check.
+		n.isHealthy = &newStatus
+		n.Unlock() // It's better to unlock as soon as the protected section is over.
+		n.healthChan <- newStatus
+	}
+	return newStatus
+}

api/client/beacon/health_test.go

@@ -0,0 +1,118 @@
+package beacon
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	healthTesting "github.com/prysmaticlabs/prysm/v5/api/client/beacon/testing"
+	"go.uber.org/mock/gomock"
+)
+
+func TestNodeHealth_IsHealthy(t *testing.T) {
+	tests := []struct {
+		name      string
+		isHealthy bool
+		want      bool
+	}{
+		{"initially healthy", true, true},
+		{"initially unhealthy", false, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			n := &NodeHealthTracker{
+				isHealthy:  &tt.isHealthy,
+				healthChan: make(chan bool, 1),
+			}
+			if got := n.IsHealthy(); got != tt.want {
+				t.Errorf("IsHealthy() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestNodeHealth_UpdateNodeHealth(t *testing.T) {
+	tests := []struct {
+		name       string
+		initial    bool // Initial health status
+		newStatus  bool // Status to update to
+		shouldSend bool // Should a message be sent through the channel
+	}{
+		{"healthy to unhealthy", true, false, true},
+		{"unhealthy to healthy", false, true, true},
+		{"remain healthy", true, true, false},
+		{"remain unhealthy", false, false, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+			client := healthTesting.NewMockHealthClient(ctrl)
+			client.EXPECT().IsHealthy(gomock.Any()).Return(tt.newStatus)
+			n := &NodeHealthTracker{
+				isHealthy:  &tt.initial,
+				node:       client,
+				healthChan: make(chan bool, 1),
+			}
+
+			s := n.CheckHealth(context.Background())
+			// Check if health status was updated
+			if s != tt.newStatus {
+				t.Errorf("UpdateNodeHealth() failed to update isHealthy from %v to %v", tt.initial, tt.newStatus)
+			}
+
+			select {
+			case status := <-n.HealthUpdates():
+				if !tt.shouldSend {
+					t.Errorf("UpdateNodeHealth() unexpectedly sent status %v to HealthCh", status)
+				} else if status != tt.newStatus {
+					t.Errorf("UpdateNodeHealth() sent wrong status %v, want %v", status, tt.newStatus)
+				}
+			default:
+				if tt.shouldSend {
+					t.Error("UpdateNodeHealth() did not send any status to HealthCh when expected")
+				}
+			}
+		})
+	}
+}
+
+func TestNodeHealth_Concurrency(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	client := healthTesting.NewMockHealthClient(ctrl)
+	n := NewNodeHealthTracker(client)
+	var wg sync.WaitGroup
+
+	// Number of goroutines to spawn for both reading and writing
+	numGoroutines := 6
+
+	go func() {
+		for range n.HealthUpdates() {
+			// Consume values to avoid blocking on channel send.
+		}
+	}()
+
+	wg.Add(numGoroutines * 2) // for readers and writers
+
+	// Concurrently update health status
+	for i := 0; i < numGoroutines; i++ {
+		go func() {
+			defer wg.Done()
+			client.EXPECT().IsHealthy(gomock.Any()).Return(false)
+			n.CheckHealth(context.Background())
+			client.EXPECT().IsHealthy(gomock.Any()).Return(true)
+			n.CheckHealth(context.Background())
+		}()
+	}
+
+	// Concurrently read health status
+	for i := 0; i < numGoroutines; i++ {
+		go func() {
+			defer wg.Done()
+			_ = n.IsHealthy() // Just read the value
+		}()
+	}
+
+	wg.Wait() // Wait for all goroutines to finish
+}

api/client/beacon/iface/BUILD.bazel

@@ -0,0 +1,8 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["health.go"],
+    importpath = "github.com/prysmaticlabs/prysm/v5/api/client/beacon/iface",
+    visibility = ["//visibility:public"],
+)

api/client/beacon/iface/health.go

@@ -0,0 +1,13 @@
+package iface
+
+import "context"
+
+type HealthTracker interface {
+	HealthUpdates() <-chan bool
+	IsHealthy() bool
+	CheckHealth(ctx context.Context) bool
+}
+
+type HealthNode interface {
+	IsHealthy(ctx context.Context) bool
+}

api/client/beacon/testing/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["mock.go"],
+    importpath = "github.com/prysmaticlabs/prysm/v5/api/client/beacon/testing",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api/client/beacon/iface:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

api/client/beacon/testing/mock.go

@@ -0,0 +1,53 @@
+package testing
+
+import (
+	"context"
+	"reflect"
+
+	"github.com/prysmaticlabs/prysm/v5/api/client/beacon/iface"
+	"go.uber.org/mock/gomock"
+)
+
+var (
+	_ = iface.HealthNode(&MockHealthClient{})
+)
+
+// MockHealthClient is a mock of HealthClient interface.
+type MockHealthClient struct {
+	ctrl     *gomock.Controller
+	recorder *MockHealthClientMockRecorder
+}
+
+// MockHealthClientMockRecorder is the mock recorder for MockHealthClient.
+type MockHealthClientMockRecorder struct {
+	mock *MockHealthClient
+}
+
+// IsHealthy mocks base method.
+func (m *MockHealthClient) IsHealthy(arg0 context.Context) bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsHealthy", arg0)
+	ret0, ok := ret[0].(bool)
+	if !ok {
+		return false
+	}
+	return ret0
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockHealthClient) EXPECT() *MockHealthClientMockRecorder {
+	return m.recorder
+}
+
+// IsHealthy indicates an expected call of IsHealthy.
+func (mr *MockHealthClientMockRecorder) IsHealthy(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsHealthy", reflect.TypeOf((*MockHealthClient)(nil).IsHealthy), arg0)
+}
+
+// NewMockHealthClient creates a new mock instance.
+func NewMockHealthClient(ctrl *gomock.Controller) *MockHealthClient {
+	mock := &MockHealthClient{ctrl: ctrl}
+	mock.recorder = &MockHealthClientMockRecorder{mock}
+	return mock
+}

api/client/errors.go

@@ -21,6 +21,9 @@ var ErrNotFound = errors.Wrap(ErrNotOK, "recv 404 NotFound response from API")
 // ErrInvalidNodeVersion indicates that the /eth/v1/node/version API response format was not recognized.
 var ErrInvalidNodeVersion = errors.New("invalid node version response")
 
+// ErrConnectionIssue represents a connection problem.
+var ErrConnectionIssue = errors.New("could not connect")
+
 // Non200Err is a function that parses an HTTP response to handle responses that are not 200 with a formatted error.
 func Non200Err(response *http.Response) error {
 	bodyBytes, err := io.ReadAll(response.Body)

api/client/event/BUILD.bazel

@@ -0,0 +1,24 @@
+load("@prysm//tools/go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["event_stream.go"],
+    importpath = "github.com/prysmaticlabs/prysm/v5/api/client/event",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api:go_default_library",
+        "//api/client:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["event_stream_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//testing/require:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)