fix: health endpoint now detects stopped block production

This commit fixes issue #2643 where the health endpoint still reports
OK when a node has stopped producing blocks.

Changes:
- Updated HealthServer to accept store, config, and logger dependencies
- Implemented block production monitoring in the Livez endpoint:
  * For aggregator nodes, checks if LastBlockTime is recent
  * Returns WARN if block production is slow (> 3x block time)
  * Returns FAIL if block production has stopped (> 5x block time)
  * Uses LazyBlockInterval for lazy mode aggregators
  * Non-aggregator nodes continue to return PASS
- Added constants for health check thresholds:
  * healthCheckWarnMultiplier = 3
  * healthCheckFailMultiplier = 5
- Added comprehensive unit tests covering all scenarios:
  Server tests (pkg/rpc/server/server_test.go):
  * Non-aggregator nodes
  * Aggregator with no blocks
  * Aggregator with recent blocks (PASS)
  * Aggregator with slow production (WARN)
  * Aggregator with stopped production (FAIL)
  * Lazy aggregator with correct thresholds
  * Error handling
  Client tests (pkg/rpc/client/client_test.go):
  * Non-aggregator returns PASS
  * Aggregator with recent blocks returns PASS
  * Aggregator with slow block production returns WARN
  * Aggregator with stopped block production returns FAIL
- Updated setupTestServer to pass new dependencies
- Added createCustomTestServer helper for testing with custom configs

The thresholds are configurable based on the node's BlockTime or
LazyBlockInterval settings, making the health check adaptive to
different node configurations.

Fixes #2643

Author: randygrok

node/helpers_test.go

@@ -24,6 +24,7 @@ import (
 	evconfig "github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/p2p"
 	"github.com/evstack/ev-node/pkg/p2p/key"
+	rpcclient "github.com/evstack/ev-node/pkg/rpc/client"
 	remote_signer "github.com/evstack/ev-node/pkg/signer/noop"
 	"github.com/evstack/ev-node/types"
 )
@@ -317,3 +318,8 @@ func verifyNodesSynced(node1, syncingNode Node, source Source) error {
 		return fmt.Errorf("nodes not synced: sequencer at height %v, syncing node at height %v", sequencerHeight, syncingHeight)
 	})
 }
+
+// NewRPCClient creates a new RPC client for testing
+func NewRPCClient(address string) *rpcclient.Client {
+	return rpcclient.NewClient("http://" + address)
+}

node/single_sequencer_integration_test.go

@@ -418,3 +418,70 @@ func waitForBlockN(t *testing.T, n uint64, node *FullNode, blockInterval time.Du
 		return got >= n
 	}, timeout[0], blockInterval/2)
 }
+// TestHealthEndpointWhenBlockProductionStops verifies that the health endpoint
+// correctly reports WARN and FAIL states when an aggregator stops producing blocks.
+func TestHealthEndpointWhenBlockProductionStops(t *testing.T) {
+	require := require.New(t)
+
+	// Set up configuration with specific block time for predictable health checks
+	config := getTestConfig(t, 1)
+	config.Node.Aggregator = true
+	config.Node.BlockTime = evconfig.DurationWrapper{Duration: 500 * time.Millisecond}
+	config.Node.MaxPendingHeadersAndData = 2
+
+	// Set DA block time large enough to avoid header submission to DA layer
+	// This will cause block production to stop once MaxPendingHeadersAndData is reached
+	config.DA.BlockTime = evconfig.DurationWrapper{Duration: 100 * time.Second}
+
+	node, cleanup := createNodeWithCleanup(t, config)
+	defer cleanup()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var runningWg sync.WaitGroup
+	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0, nil)
+
+	// Wait for first block to be produced
+	waitForBlockN(t, 1, node, config.Node.BlockTime.Duration)
+
+	// Create RPC client
+	rpcClient := NewRPCClient(config.RPC.Address)
+
+	// Verify health is PASS while blocks are being produced
+	health, err := rpcClient.GetHealth(ctx)
+	require.NoError(err)
+	require.Equal("PASS", health.String(), "Health should be PASS while producing blocks")
+
+	// Wait for block production to stop (when MaxPendingHeadersAndData is reached)
+	time.Sleep(time.Duration(config.Node.MaxPendingHeadersAndData+2) * config.Node.BlockTime.Duration)
+
+	// Get the height to confirm blocks stopped
+	height, err := getNodeHeight(node, Store)
+	require.NoError(err)
+	require.LessOrEqual(height, config.Node.MaxPendingHeadersAndData)
+
+	// Health check threshold calculations:
+	// blockTime = 500ms
+	// warnThreshold = blockTime * 3 = 1500ms = 1.5s
+	// failThreshold = blockTime * 5 = 2500ms = 2.5s
+
+	// Wait for WARN threshold (3x block time = 1.5 seconds after last block)
+	// We need to wait a bit longer to account for the time blocks take to stop
+	time.Sleep(1700 * time.Millisecond)
+
+	health, err = rpcClient.GetHealth(ctx)
+	require.NoError(err)
+	// Health could be WARN or FAIL depending on exact timing, but should not be PASS
+	require.NotEqual("PASS", health.String(), "Health should not be PASS after block production stops")
+
+	// Wait for FAIL threshold (5x block time = 2.5 seconds total after last block)
+	time.Sleep(1500 * time.Millisecond)
+
+	health, err = rpcClient.GetHealth(ctx)
+	require.NoError(err)
+	require.Equal("FAIL", health.String(), "Health should be FAIL after 5x block time without new blocks")
+
+	// Stop the node and wait for shutdown
+	shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second)
+}

pkg/rpc/client/client_test.go

@@ -30,13 +30,12 @@ func setupTestServer(t *testing.T, mockStore *mocks.MockStore, mockP2P *mocks.Mo
 
 	// Create the servers
 	logger := zerolog.Nop()
-	storeServer := server.NewStoreServer(mockStore, logger)
-	p2pServer := server.NewP2PServer(mockP2P)
-	healthServer := server.NewHealthServer()
-
-	// Create config server with test config
 	testConfig := config.DefaultConfig()
 	testConfig.DA.Namespace = "test-headers"
+
+	storeServer := server.NewStoreServer(mockStore, logger)
+	p2pServer := server.NewP2PServer(mockP2P)
+	healthServer := server.NewHealthServer(mockStore, testConfig, logger)
 	configServer := server.NewConfigServer(testConfig, nil, logger)
 
 	// Register the store service
@@ -242,21 +241,123 @@ func TestClientGetNetInfo(t *testing.T) {
 }
 
 func TestClientGetHealth(t *testing.T) {
-	// Create mocks
-	mockStore := mocks.NewMockStore(t)
-	mockP2P := mocks.NewMockP2PRPC(t)
+	t.Run("non-aggregator returns PASS", func(t *testing.T) {
+		mockStore := mocks.NewMockStore(t)
+		mockP2P := mocks.NewMockP2PRPC(t)
+
+		testServer, client := setupTestServer(t, mockStore, mockP2P)
+		defer testServer.Close()
+
+		healthStatus, err := client.GetHealth(context.Background())
+
+		require.NoError(t, err)
+		require.Equal(t, "PASS", healthStatus.String())
+	})
+
+	t.Run("aggregator with recent blocks returns PASS", func(t *testing.T) {
+		mockStore := mocks.NewMockStore(t)
+		mockP2P := mocks.NewMockP2PRPC(t)
+
+		// Setup aggregator config
+		testConfig := config.DefaultConfig()
+		testConfig.Node.Aggregator = true
+		testConfig.Node.BlockTime.Duration = 1 * time.Second
+
+		// Create state with recent block
+		state := types.State{
+			LastBlockHeight: 100,
+			LastBlockTime:   time.Now().Add(-500 * time.Millisecond),
+		}
+		mockStore.On("GetState", mock.Anything).Return(state, nil)
+
+		// Create custom test server with aggregator config
+		testServer := createCustomTestServer(t, mockStore, mockP2P, testConfig)
+		defer testServer.Close()
+
+		client := NewClient(testServer.URL)
+		healthStatus, err := client.GetHealth(context.Background())
+
+		require.NoError(t, err)
+		require.Equal(t, "PASS", healthStatus.String())
+		mockStore.AssertExpectations(t)
+	})
+
+	t.Run("aggregator with slow block production returns WARN", func(t *testing.T) {
+		mockStore := mocks.NewMockStore(t)
+		mockP2P := mocks.NewMockP2PRPC(t)
+
+		testConfig := config.DefaultConfig()
+		testConfig.Node.Aggregator = true
+		testConfig.Node.BlockTime.Duration = 1 * time.Second
+
+		// State with block older than 3x block time
+		state := types.State{
+			LastBlockHeight: 100,
+			LastBlockTime:   time.Now().Add(-4 * time.Second),
+		}
+		mockStore.On("GetState", mock.Anything).Return(state, nil)
+
+		testServer := createCustomTestServer(t, mockStore, mockP2P, testConfig)
+		defer testServer.Close()
+
+		client := NewClient(testServer.URL)
+		healthStatus, err := client.GetHealth(context.Background())
+
+		require.NoError(t, err)
+		require.Equal(t, "WARN", healthStatus.String())
+		mockStore.AssertExpectations(t)
+	})
+
+	t.Run("aggregator with stopped block production returns FAIL", func(t *testing.T) {
+		mockStore := mocks.NewMockStore(t)
+		mockP2P := mocks.NewMockP2PRPC(t)
+
+		testConfig := config.DefaultConfig()
+		testConfig.Node.Aggregator = true
+		testConfig.Node.BlockTime.Duration = 1 * time.Second
+
+		// State with block older than 5x block time
+		state := types.State{
+			LastBlockHeight: 100,
+			LastBlockTime:   time.Now().Add(-10 * time.Second),
+		}
+		mockStore.On("GetState", mock.Anything).Return(state, nil)
+
+		testServer := createCustomTestServer(t, mockStore, mockP2P, testConfig)
+		defer testServer.Close()
+
+		client := NewClient(testServer.URL)
+		healthStatus, err := client.GetHealth(context.Background())
+
+		require.NoError(t, err)
+		require.Equal(t, "FAIL", healthStatus.String())
+		mockStore.AssertExpectations(t)
+	})
+}
 
-	// Setup test server and client
-	testServer, client := setupTestServer(t, mockStore, mockP2P)
-	defer testServer.Close()
+// createCustomTestServer creates a test server with custom configuration
+func createCustomTestServer(t *testing.T, mockStore *mocks.MockStore, mockP2P *mocks.MockP2PRPC, testConfig config.Config) *httptest.Server {
+	mux := http.NewServeMux()
+
+	logger := zerolog.Nop()
+	storeServer := server.NewStoreServer(mockStore, logger)
+	p2pServer := server.NewP2PServer(mockP2P)
+	healthServer := server.NewHealthServer(mockStore, testConfig, logger)
+	configServer := server.NewConfigServer(testConfig, nil, logger)
 
-	// Call GetHealth
-	healthStatus, err := client.GetHealth(context.Background())
+	storePath, storeHandler := rpc.NewStoreServiceHandler(storeServer)
+	mux.Handle(storePath, storeHandler)
 
-	// Assert expectations
-	require.NoError(t, err)
-	// Health server always returns PASS in Livez
-	require.NotEqual(t, healthStatus.String(), "UNKNOWN")
+	p2pPath, p2pHandler := rpc.NewP2PServiceHandler(p2pServer)
+	mux.Handle(p2pPath, p2pHandler)
+
+	healthPath, healthHandler := rpc.NewHealthServiceHandler(healthServer)
+	mux.Handle(healthPath, healthHandler)
+
+	configPath, configHandler := rpc.NewConfigServiceHandler(configServer)
+	mux.Handle(configPath, configHandler)
+
+	return httptest.NewServer(h2c.NewHandler(mux, &http2.Server{}))
 }
 
 func TestClientGetNamespace(t *testing.T) {

pkg/rpc/server/server.go

@@ -28,6 +28,16 @@ import (
 	rpc "github.com/evstack/ev-node/types/pb/evnode/v1/v1connect"
 )
 
+const (
+	// healthCheckWarnMultiplier is the multiplier for block time to determine WARN threshold
+	// If no block has been produced in (blockTime * healthCheckWarnMultiplier), return WARN
+	healthCheckWarnMultiplier = 3
+
+	// healthCheckFailMultiplier is the multiplier for block time to determine FAIL threshold
+	// If no block has been produced in (blockTime * healthCheckFailMultiplier), return FAIL
+	healthCheckFailMultiplier = 5
+)
+
 var _ rpc.StoreServiceHandler = (*StoreServer)(nil)
 
 // StoreServer implements the StoreService defined in the proto file
@@ -288,19 +298,76 @@ func (p *P2PServer) GetNetInfo(
 }
 
 // HealthServer implements the HealthService defined in the proto file
-type HealthServer struct{}
+type HealthServer struct {
+	store  store.Store
+	config config.Config
+	logger zerolog.Logger
+}
 
 // NewHealthServer creates a new HealthServer instance
-func NewHealthServer() *HealthServer {
-	return &HealthServer{}
+func NewHealthServer(store store.Store, config config.Config, logger zerolog.Logger) *HealthServer {
+	return &HealthServer{
+		store:  store,
+		config: config,
+		logger: logger,
+	}
 }
 
 // Livez implements the HealthService.Livez RPC
 func (h *HealthServer) Livez(
 	ctx context.Context,
 	req *connect.Request[emptypb.Empty],
 ) (*connect.Response[pb.GetHealthResponse], error) {
-	// always return healthy
+	// For aggregator nodes, check if block production is healthy
+	if h.config.Node.Aggregator {
+		state, err := h.store.GetState(ctx)
+		if err != nil {
+			h.logger.Error().Err(err).Msg("Failed to get state for health check")
+			return connect.NewResponse(&pb.GetHealthResponse{
+				Status: pb.HealthStatus_FAIL,
+			}), nil
+		}
+
+		// If we have blocks, check if the last block time is recent
+		if state.LastBlockHeight > 0 {
+			timeSinceLastBlock := time.Since(state.LastBlockTime)
+
+			// Calculate the threshold based on block time
+			blockTime := h.config.Node.BlockTime.Duration
+
+			// For lazy mode, use the lazy block interval instead
+			if h.config.Node.LazyMode {
+				blockTime = h.config.Node.LazyBlockInterval.Duration
+			}
+
+			warnThreshold := blockTime * healthCheckWarnMultiplier
+			failThreshold := blockTime * healthCheckFailMultiplier
+
+			if timeSinceLastBlock > failThreshold {
+				h.logger.Warn().
+					Dur("time_since_last_block", timeSinceLastBlock).
+					Dur("fail_threshold", failThreshold).
+					Uint64("last_block_height", state.LastBlockHeight).
+					Time("last_block_time", state.LastBlockTime).
+					Msg("Health check: node has stopped producing blocks (FAIL)")
+				return connect.NewResponse(&pb.GetHealthResponse{
+					Status: pb.HealthStatus_FAIL,
+				}), nil
+			} else if timeSinceLastBlock > warnThreshold {
+				h.logger.Warn().
+					Dur("time_since_last_block", timeSinceLastBlock).
+					Dur("warn_threshold", warnThreshold).
+					Uint64("last_block_height", state.LastBlockHeight).
+					Time("last_block_time", state.LastBlockTime).
+					Msg("Health check: block production is slow (WARN)")
+				return connect.NewResponse(&pb.GetHealthResponse{
+					Status: pb.HealthStatus_WARN,
+				}), nil
+			}
+		}
+	}
+
+	// For non-aggregator nodes or if checks pass, return healthy
 	return connect.NewResponse(&pb.GetHealthResponse{
 		Status: pb.HealthStatus_PASS,
 	}), nil
@@ -310,7 +377,7 @@ func (h *HealthServer) Livez(
 func NewServiceHandler(store store.Store, peerManager p2p.P2PRPC, proposerAddress []byte, logger zerolog.Logger, config config.Config, bestKnown BestKnownHeightProvider) (http.Handler, error) {
 	storeServer := NewStoreServer(store, logger)
 	p2pServer := NewP2PServer(peerManager)
-	healthServer := NewHealthServer()
+	healthServer := NewHealthServer(store, config, logger)
 	configServer := NewConfigServer(config, proposerAddress, logger)
 
 	mux := http.NewServeMux()