feat: add /live and /ready endpoints with support during DB migration (#2923)

- Add /live and /ready endpoints for Kubernetes liveness and readiness probes
- Enable HTTP status server during database migrations to serve health checks
- Include descriptive response messages for migration and sync status

This allows Kubernetes to monitor node health even during database migrations,
preventing unnecessary pod restarts during migration processes.

Author: wojciechos

migration/migration.go

@@ -8,8 +8,12 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"net"
+	"net/http"
 	"runtime"
+	"strconv"
 	"sync"
+	"time"
 
 	"github.com/NethermindEth/juno/adapters/sn2core"
 	"github.com/NethermindEth/juno/core"
@@ -29,6 +33,12 @@ type schemaMetadata struct {
 	IntermediateState []byte
 }
 
+type HTTPConfig struct {
+	Enabled bool
+	Host    string
+	Port    uint16
+}
+
 type Migration interface {
 	Before(intermediateState []byte) error
 	// Migration should return intermediate state whenever it requests new txn or detects cancelled ctx.
@@ -76,8 +86,10 @@ var defaultMigrations = []Migration{
 
 var ErrCallWithNewTransaction = errors.New("call with new transaction")
 
-func MigrateIfNeeded(ctx context.Context, targetDB db.KeyValueStore, network *utils.Network, log utils.SimpleLogger) error {
-	return migrateIfNeeded(ctx, targetDB, network, log, defaultMigrations)
+func MigrateIfNeeded(ctx context.Context, targetDB db.KeyValueStore, network *utils.Network,
+	log utils.SimpleLogger, httpConfig *HTTPConfig,
+) error {
+	return migrateIfNeeded(ctx, targetDB, network, log, defaultMigrations, httpConfig)
 }
 
 func migrateIfNeeded(
@@ -86,6 +98,7 @@ func migrateIfNeeded(
 	network *utils.Network,
 	log utils.SimpleLogger,
 	migrations []Migration,
+	httpConfig *HTTPConfig,
 ) error {
 	/*
 		Schema metadata of the targetDB determines which set of migrations need to be applied to the database.
@@ -113,6 +126,11 @@ func migrateIfNeeded(
 		return errors.New("db is from a newer, incompatible version of Juno; upgrade to use this database")
 	}
 
+	if httpConfig.Enabled {
+		migrationSrv := startMigrationStatusServer(log, httpConfig.Host, httpConfig.Port)
+		defer closeMigrationServer(migrationSrv, log)
+	}
+
 	for i := metadata.Version; i < currentVersion; i++ {
 		if err = ctx.Err(); err != nil {
 			return err
@@ -155,6 +173,13 @@ func migrateIfNeeded(
 	return nil
 }
 
+func closeMigrationServer(srv *http.Server, log utils.SimpleLogger) {
+	log.Debugw("Closing migration status server immediately...")
+	if err := srv.Close(); err != nil {
+		log.Errorw("Migration status server close failed", "err", err)
+	}
+}
+
 // SchemaMetadata retrieves metadata about a database schema from the given database.
 func SchemaMetadata(targetDB db.KeyValueStore) (schemaMetadata, error) {
 	metadata := schemaMetadata{}
@@ -869,3 +894,35 @@ func reconstructAggregatedBloomFilters(txn db.IndexedBatch, network *utils.Netwo
 
 	return core.WriteRunningEventFilter(txn, runningFilter)
 }
+
+func startMigrationStatusServer(log utils.SimpleLogger, host string, port uint16) *http.Server {
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("/live", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+
+	mux.HandleFunc("/ready", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		_, err := w.Write([]byte("Database migration in progress."))
+		if err != nil {
+			log.Errorw("Failed to write migration status response", "err", err)
+		}
+	})
+
+	portStr := strconv.FormatUint(uint64(port), 10)
+	addr := net.JoinHostPort(host, portStr)
+	srv := &http.Server{
+		Addr:              addr,
+		Handler:           mux,
+		ReadHeaderTimeout: 30 * time.Second,
+	}
+
+	go func() {
+		log.Debugw("Starting migration status server on " + addr)
+		if err := srv.ListenAndServe(); err != http.ErrServerClosed {
+			log.Errorw("Migration status server failed", "err", err)
+		}
+	}()
+	return srv
+}

migration/migration_pkg_test.go

@@ -502,7 +502,7 @@ func TestMigrateIfNeeded(t *testing.T) {
 				},
 			},
 		}
-		require.ErrorContains(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations), "bar")
+		require.ErrorContains(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations, &HTTPConfig{}), "bar")
 	})
 
 	t.Run("call with new tx", func(t *testing.T) {
@@ -522,7 +522,7 @@ func TestMigrateIfNeeded(t *testing.T) {
 				},
 			},
 		}
-		require.NoError(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations))
+		require.NoError(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations, &HTTPConfig{}))
 	})
 
 	t.Run("error during migration", func(t *testing.T) {
@@ -537,7 +537,7 @@ func TestMigrateIfNeeded(t *testing.T) {
 				},
 			},
 		}
-		require.ErrorContains(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations), "foo")
+		require.ErrorContains(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations, &HTTPConfig{}), "foo")
 	})
 
 	t.Run("error if using new db on old version of juno", func(t *testing.T) {
@@ -552,9 +552,9 @@ func TestMigrateIfNeeded(t *testing.T) {
 				},
 			},
 		}
-		require.NoError(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations))
+		require.NoError(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), migrations, &HTTPConfig{}))
 		want := "db is from a newer, incompatible version of Juno"
-		require.ErrorContains(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), []Migration{}), want)
+		require.ErrorContains(t, migrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), []Migration{}, &HTTPConfig{}), want)
 	})
 }
 

migration/migration_test.go

@@ -16,7 +16,7 @@ func TestMigrateIfNeeded(t *testing.T) {
 	ctx, cancel := context.WithCancel(t.Context())
 	cancel()
 	t.Run("Migration should not happen on cancelled ctx", func(t *testing.T) {
-		require.ErrorIs(t, migration.MigrateIfNeeded(ctx, testDB, &utils.Mainnet, utils.NewNopZapLogger()), ctx.Err())
+		require.ErrorIs(t, migration.MigrateIfNeeded(ctx, testDB, &utils.Mainnet, utils.NewNopZapLogger(), &migration.HTTPConfig{}), ctx.Err())
 	})
 
 	meta, err := migration.SchemaMetadata(testDB)
@@ -25,7 +25,7 @@ func TestMigrateIfNeeded(t *testing.T) {
 	require.Nil(t, meta.IntermediateState)
 
 	t.Run("Migration should happen on empty DB", func(t *testing.T) {
-		require.NoError(t, migration.MigrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger()))
+		require.NoError(t, migration.MigrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), &migration.HTTPConfig{}))
 	})
 
 	meta, err = migration.SchemaMetadata(testDB)
@@ -34,7 +34,7 @@ func TestMigrateIfNeeded(t *testing.T) {
 	require.Nil(t, meta.IntermediateState)
 
 	t.Run("subsequent calls to MigrateIfNeeded should not change the DB version", func(t *testing.T) {
-		require.NoError(t, migration.MigrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger()))
+		require.NoError(t, migration.MigrateIfNeeded(t.Context(), testDB, &utils.Mainnet, utils.NewNopZapLogger(), &migration.HTTPConfig{}))
 		postVersion, postErr := migration.SchemaMetadata(testDB)
 		require.NoError(t, postErr)
 		require.Equal(t, meta, postVersion)

node/http.go

@@ -239,6 +239,7 @@ func NewReadinessHandlers(bcReader blockchain.Reader, syncReader sync.Reader) *r
 func (h *readinessHandlers) HandleReadySync(w http.ResponseWriter, r *http.Request) {
 	if !h.isSynced() {
 		w.WriteHeader(http.StatusServiceUnavailable)
+		w.Write([]byte("Node not synced yet.")) //nolint:errcheck
 		return
 	}
 
@@ -261,3 +262,7 @@ func (h *readinessHandlers) isSynced() bool {
 
 	return head.Number+SyncBlockRange >= highestBlockHeader.Number
 }
+
+func (h *readinessHandlers) HandleLive(w http.ResponseWriter, r *http.Request) {
+	w.WriteHeader(http.StatusOK)
+}

node/node.go

@@ -285,6 +285,8 @@ func New(cfg *Config, version string, logLevel *utils.LogLevel) (*Node, error) {
 	if cfg.HTTP {
 		readinessHandlers := NewReadinessHandlers(chain, synchronizer)
 		httpHandlers := map[string]http.HandlerFunc{
+			"/live":       readinessHandlers.HandleLive,
+			"/ready":      readinessHandlers.HandleReadySync,
 			"/ready/sync": readinessHandlers.HandleReadySync,
 		}
 		services = append(services, makeRPCOverHTTP(cfg.HTTPHost, cfg.HTTPPort, rpcServers, httpHandlers, log, cfg.Metrics, cfg.RPCCorsEnable))
@@ -428,7 +430,13 @@ func (n *Node) Run(ctx context.Context) {
 		n.StartService(wg, ctx, cancel, s)
 	}
 
-	if err := migration.MigrateIfNeeded(ctx, n.db, &n.cfg.Network, n.log); err != nil {
+	migrationHTTPConfig := migration.HTTPConfig{
+		Enabled: n.cfg.HTTP,
+		Host:    n.cfg.HTTPHost,
+		Port:    n.cfg.HTTPPort,
+	}
+
+	if err := migration.MigrateIfNeeded(ctx, n.db, &n.cfg.Network, n.log, &migrationHTTPConfig); err != nil {
 		if errors.Is(err, context.Canceled) {
 			n.log.Infow("DB Migration cancelled")
 			return