Merge branch 'main' into marko/2603

Author: tac0turtle

node/full.go

@@ -281,7 +281,13 @@ func (n *FullNode) Run(parentCtx context.Context) error {
 	}
 
 	// Start RPC server
-	handler, err := rpcserver.NewServiceHandler(n.Store, n.p2pClient, n.genesis.ProposerAddress, n.Logger, n.nodeConfig)
+	bestKnownHeightProvider := func() uint64 {
+		hHeight := n.hSyncService.Store().Height()
+		dHeight := n.dSyncService.Store().Height()
+		return min(hHeight, dHeight)
+	}
+
+	handler, err := rpcserver.NewServiceHandler(n.Store, n.p2pClient, n.genesis.ProposerAddress, n.Logger, n.nodeConfig, bestKnownHeightProvider)
 	if err != nil {
 		return fmt.Errorf("error creating RPC handler: %w", err)
 	}

node/light.go

@@ -77,7 +77,13 @@ func (ln *LightNode) Run(parentCtx context.Context) error {
 
 	ln.running = true
 	// Start RPC server (light node doesn't have a signer)
-	handler, err := rpcserver.NewServiceHandler(ln.Store, ln.P2P, nil, ln.Logger, ln.nodeConfig)
+
+	// for light nodes use the header sync service height as the best known height.
+	bestKnown := func() uint64 {
+		return ln.hSyncService.Store().Height()
+	}
+
+	handler, err := rpcserver.NewServiceHandler(ln.Store, ln.P2P, nil, ln.Logger, ln.nodeConfig, bestKnown)
 	if err != nil {
 		return fmt.Errorf("error creating RPC handler: %w", err)
 	}

pkg/config/config.go

@@ -45,6 +45,8 @@ const (
 	FlagMaxPendingHeadersAndData = FlagPrefixEvnode + "node.max_pending_headers_and_data"
 	// FlagLazyBlockTime is a flag for specifying the maximum interval between blocks in lazy aggregation mode
 	FlagLazyBlockTime = FlagPrefixEvnode + "node.lazy_block_interval"
+	// FlagReadinessMaxBlocksBehind configures how many blocks behind best-known head is still considered ready
+	FlagReadinessMaxBlocksBehind = FlagPrefixEvnode + "node.readiness_max_blocks_behind"
 
 	// Data Availability configuration flags
 
@@ -193,6 +195,9 @@ type NodeConfig struct {
 
 	// Header configuration
 	TrustedHash string `mapstructure:"trusted_hash" yaml:"trusted_hash" comment:"Initial trusted hash used to bootstrap the header exchange service. Allows nodes to start synchronizing from a specific trusted point in the chain instead of genesis. When provided, the node will fetch the corresponding header/block from peers using this hash and use it as a starting point for synchronization. If not provided, the node will attempt to fetch the genesis block instead."`
+
+	// Readiness / health configuration
+	ReadinessMaxBlocksBehind uint64 `mapstructure:"readiness_max_blocks_behind" yaml:"readiness_max_blocks_behind" comment:"How many blocks behind best-known head the node can be and still be considered ready. 0 means must be exactly at head."`
 }
 
 // LogConfig contains all logging configuration parameters
@@ -307,6 +312,7 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool(FlagLazyAggregator, def.Node.LazyMode, "produce blocks only when transactions are available or after lazy block time")
 	cmd.Flags().Uint64(FlagMaxPendingHeadersAndData, def.Node.MaxPendingHeadersAndData, "maximum headers or data pending DA confirmation before pausing block production (0 for no limit)")
 	cmd.Flags().Duration(FlagLazyBlockTime, def.Node.LazyBlockInterval.Duration, "maximum interval between blocks in lazy aggregation mode")
+	cmd.Flags().Uint64(FlagReadinessMaxBlocksBehind, def.Node.ReadinessMaxBlocksBehind, "how many blocks behind best-known head the node can be and still be considered ready (0 = must be at head)")
 
 	// Data Availability configuration flags
 	cmd.Flags().String(FlagDAAddress, def.DA.Address, "DA address (host:port)")

pkg/config/config_test.go

@@ -63,6 +63,7 @@ func TestAddFlags(t *testing.T) {
 	assertFlagValue(t, flags, FlagLazyAggregator, DefaultConfig().Node.LazyMode)
 	assertFlagValue(t, flags, FlagMaxPendingHeadersAndData, DefaultConfig().Node.MaxPendingHeadersAndData)
 	assertFlagValue(t, flags, FlagLazyBlockTime, DefaultConfig().Node.LazyBlockInterval.Duration)
+	assertFlagValue(t, flags, FlagReadinessMaxBlocksBehind, DefaultConfig().Node.ReadinessMaxBlocksBehind)
 
 	// DA flags
 	assertFlagValue(t, flags, FlagDAAddress, DefaultConfig().DA.Address)
@@ -104,7 +105,7 @@ func TestAddFlags(t *testing.T) {
 	assertFlagValue(t, flags, FlagRPCAddress, DefaultConfig().RPC.Address)
 
 	// Count the number of flags we're explicitly checking
-	expectedFlagCount := 37 // Update this number if you add more flag checks above
+	expectedFlagCount := 38 // Update this number if you add more flag checks above
 
 	// Get the actual number of flags (both regular and persistent)
 	actualFlagCount := 0

pkg/config/defaults.go

@@ -46,12 +46,13 @@ func DefaultConfig() Config {
 			Peers:         "",
 		},
 		Node: NodeConfig{
-			Aggregator:        false,
-			BlockTime:         DurationWrapper{1 * time.Second},
-			LazyMode:          false,
-			LazyBlockInterval: DurationWrapper{60 * time.Second},
-			Light:             false,
-			TrustedHash:       "",
+			Aggregator:               false,
+			BlockTime:                DurationWrapper{1 * time.Second},
+			LazyMode:                 false,
+			LazyBlockInterval:        DurationWrapper{60 * time.Second},
+			Light:                    false,
+			TrustedHash:              "",
+			ReadinessMaxBlocksBehind: 3,
 		},
 		DA: DAConfig{
 			Address:           "http://localhost:7980",

pkg/rpc/example/example.go

@@ -20,8 +20,9 @@ func StartStoreServer(s store.Store, address string, logger zerolog.Logger) {
 	// Create and start the server
 	// Start RPC server
 	rpcAddr := fmt.Sprintf("%s:%d", "localhost", 8080)
+
 	cfg := config.DefaultConfig()
-	handler, err := server.NewServiceHandler(s, nil, nil, logger, cfg)
+	handler, err := server.NewServiceHandler(s, nil, nil, logger, cfg, nil)
 	if err != nil {
 		panic(err)
 	}
@@ -81,7 +82,7 @@ func ExampleServer(s store.Store) {
 	// Start RPC server
 	rpcAddr := fmt.Sprintf("%s:%d", "localhost", 8080)
 	cfg := config.DefaultConfig()
-	handler, err := server.NewServiceHandler(s, nil, nil, logger, cfg)
+	handler, err := server.NewServiceHandler(s, nil, nil, logger, cfg, nil)
 	if err != nil {
 		panic(err)
 	}

pkg/rpc/server/da_visualization_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	coreda "github.com/evstack/ev-node/core/da"
+	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/test/mocks"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
@@ -254,7 +255,7 @@ func TestRegisterCustomHTTPEndpointsDAVisualization(t *testing.T) {
 
 	// Create mux and register endpoints
 	mux := http.NewServeMux()
-	RegisterCustomHTTPEndpoints(mux)
+	RegisterCustomHTTPEndpoints(mux, nil, nil, config.DefaultConfig(), nil)
 
 	// Test /da endpoint
 	req, err := http.NewRequest("GET", "/da", nil)
@@ -291,7 +292,7 @@ func TestRegisterCustomHTTPEndpointsWithoutServer(t *testing.T) {
 	SetDAVisualizationServer(nil)
 
 	mux := http.NewServeMux()
-	RegisterCustomHTTPEndpoints(mux)
+	RegisterCustomHTTPEndpoints(mux, nil, nil, config.DefaultConfig(), nil)
 
 	// Test that endpoints return service unavailable when server is not set
 	endpoints := []string{"/da", "/da/submissions", "/da/blob"}

pkg/rpc/server/http.go

@@ -3,17 +3,83 @@ package server
 import (
 	"fmt"
 	"net/http"
+
+	"github.com/evstack/ev-node/pkg/config"
+	"github.com/evstack/ev-node/pkg/p2p"
+	"github.com/evstack/ev-node/pkg/store"
 )
 
+// BestKnownHeightProvider should return the best-known network height observed by the node
+// (e.g. min(headerSyncHeight, dataSyncHeight) for full nodes, or header height for light nodes).
+type BestKnownHeightProvider func() uint64
+
 // RegisterCustomHTTPEndpoints is the designated place to add new, non-gRPC, plain HTTP handlers.
 // Additional custom HTTP endpoints can be registered on the mux here.
-func RegisterCustomHTTPEndpoints(mux *http.ServeMux) {
+func RegisterCustomHTTPEndpoints(mux *http.ServeMux, s store.Store, pm p2p.P2PRPC, cfg config.Config, bestKnownHeightProvider BestKnownHeightProvider) {
 	mux.HandleFunc("/health/live", func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "text/plain")
 		w.WriteHeader(http.StatusOK)
 		fmt.Fprintln(w, "OK")
 	})
 
+	// Readiness endpoint
+	mux.HandleFunc("/health/ready", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/plain")
+
+		// Peer readiness: non-aggregator nodes should have at least 1 peer
+		if pm != nil && !cfg.Node.Aggregator {
+			peers, err := pm.GetPeers()
+			if err != nil {
+				http.Error(w, "UNREADY: failed to query peers", http.StatusServiceUnavailable)
+				return
+			}
+			if len(peers) == 0 {
+				http.Error(w, "UNREADY: no peers connected", http.StatusServiceUnavailable)
+				return
+			}
+		}
+
+		localHeight, err := s.Height(r.Context())
+		if err != nil {
+			http.Error(w, "UNREADY: state unavailable", http.StatusServiceUnavailable)
+			return
+		}
+
+		// If no blocks yet, consider unready
+		if localHeight == 0 {
+			http.Error(w, "UNREADY: no blocks yet", http.StatusServiceUnavailable)
+			return
+		}
+
+		// Require best-known height to make the readiness decision
+		if bestKnownHeightProvider == nil {
+			http.Error(w, "UNREADY: best-known height unavailable", http.StatusServiceUnavailable)
+			return
+		}
+
+		bestKnownHeight := bestKnownHeightProvider()
+		if bestKnownHeight == 0 {
+			http.Error(w, "UNREADY: best-known height unknown", http.StatusServiceUnavailable)
+			return
+		}
+
+		allowedBlocksBehind := cfg.Node.ReadinessMaxBlocksBehind
+		if bestKnownHeight <= localHeight {
+			// local is ahead of our observed best-known consider ready
+			w.WriteHeader(http.StatusOK)
+			fmt.Fprintln(w, "READY")
+			return
+		}
+
+		if bestKnownHeight-localHeight > allowedBlocksBehind {
+			http.Error(w, "UNREADY: behind best-known head", http.StatusServiceUnavailable)
+			return
+		}
+
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintln(w, "READY")
+	})
+
 	// DA Visualization endpoints
 	mux.HandleFunc("/da", func(w http.ResponseWriter, r *http.Request) {
 		server := GetDAVisualizationServer()

pkg/rpc/server/http_test.go

@@ -6,6 +6,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/evstack/ev-node/pkg/config"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -14,7 +15,7 @@ func TestRegisterCustomHTTPEndpoints(t *testing.T) {
 	mux := http.NewServeMux()
 
 	// Register custom HTTP endpoints
-	RegisterCustomHTTPEndpoints(mux)
+	RegisterCustomHTTPEndpoints(mux, nil, nil, config.DefaultConfig(), nil)
 
 	// Create a new HTTP test server with the mux
 	testServer := httptest.NewServer(mux)

pkg/rpc/server/server.go

@@ -308,7 +308,7 @@ func (h *HealthServer) Livez(
 }
 
 // NewServiceHandler creates a new HTTP handler for Store, P2P and Health services
-func NewServiceHandler(store store.Store, peerManager p2p.P2PRPC, proposerAddress []byte, logger zerolog.Logger, config config.Config) (http.Handler, error) {
+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()
@@ -342,7 +342,7 @@ func NewServiceHandler(store store.Store, peerManager p2p.P2PRPC, proposerAddres
 	mux.Handle(configPath, configHandler)
 
 	// Register custom HTTP endpoints
-	RegisterCustomHTTPEndpoints(mux)
+	RegisterCustomHTTPEndpoints(mux, store, peerManager, config, bestKnown)
 
 	// Use h2c to support HTTP/2 without TLS
 	return h2c.NewHandler(mux, &http2.Server{