Rolling updates for versiond-managed binaries (blue/green + drain, part 1 of #1367)

devshard/chainoracle/server/http.go

@@ -3,6 +3,7 @@
 package server
 
 import (
+	"context"
 	"net/http"
 
 	"devshard/chainoracle/blocks"
@@ -24,11 +25,27 @@ type VersionConfig struct {
 	Versions []Version `json:"versions"`
 }
 
+// VersionProvider returns the currently approved versions.
+type VersionProvider interface {
+	Versions(context.Context) ([]Version, error)
+}
+
+// VersionProviderFunc adapts a function to VersionProvider.
+type VersionProviderFunc func(context.Context) ([]Version, error)
+
+// Versions calls f(ctx).
+func (f VersionProviderFunc) Versions(ctx context.Context) ([]Version, error) {
+	return f(ctx)
+}
+
 // Config wires the HTTP mux.
 type Config struct {
 	Blocks blocks.BlockOracle
 	// Versions is served at GET /versions for VERSIOND_ORACLE_URL polling.
 	Versions []Version
+	// VersionProvider, when set, overrides Versions for dynamic tests and dapi
+	// implementations.
+	VersionProvider VersionProvider
 }
 
 // Mount registers chainoracle HTTP routes on g.
@@ -40,11 +57,25 @@ func Mount(g *echo.Group, cfg Config) {
 		panic("chainoracle/server: Blocks oracle is required")
 	}
 	blockserver.Mount(g, cfg.Blocks)
-	g.GET("/versions", handleVersions(cfg.Versions))
+	if cfg.VersionProvider != nil {
+		g.GET("/versions", handleVersionProvider(cfg.VersionProvider))
+	} else {
+		g.GET("/versions", handleVersions(cfg.Versions))
+	}
 }
 
 func handleVersions(versions []Version) echo.HandlerFunc {
 	return func(c echo.Context) error {
 		return c.JSON(http.StatusOK, VersionConfig{Versions: versions})
 	}
 }
+
+func handleVersionProvider(provider VersionProvider) echo.HandlerFunc {
+	return func(c echo.Context) error {
+		versions, err := provider.Versions(c.Request().Context())
+		if err != nil {
+			return echo.NewHTTPError(http.StatusBadGateway, err.Error())
+		}
+		return c.JSON(http.StatusOK, VersionConfig{Versions: versions})
+	}
+}

devshard/cmd/devshardd/app.go

@@ -30,10 +30,14 @@ import (
 const sessionEpochRetain = 3
 
 type devshardApp struct {
-	server      *echo.Echo
-	chainEvents *chainEventBridge
-	port        int
-	close       func()
+	server        *echo.Echo
+	adminServer   *echo.Echo
+	adminAddr     string
+	chainEvents   *chainEventBridge
+	port          int
+	lifecycle     *lifecycleState
+	shutdownGrace time.Duration
+	close         func()
 }
 
 type chainRuntime struct {
@@ -101,14 +105,24 @@ func buildApp(ctx context.Context, cfg runtimeConfig) (_ *devshardApp, err error
 		return nil, err
 	}
 
-	e := buildServer()
+	lifecycle := newLifecycleState()
+	e := buildServer(lifecycle)
+	var admin *echo.Echo
+	if cfg.AdminAddr != "" {
+		admin = buildAdminServer(lifecycle)
+	}
 	manager.Register(e.Group(""))
+	chainRuntime.chainEvents.OnReady(lifecycle.SetReady)
 
 	return &devshardApp{
-		server:      e,
-		chainEvents: chainRuntime.chainEvents,
-		port:        cfg.Port,
-		close:       closers.Close,
+		server:        e,
+		adminServer:   admin,
+		adminAddr:     cfg.AdminAddr,
+		chainEvents:   chainRuntime.chainEvents,
+		port:          cfg.Port,
+		lifecycle:     lifecycle,
+		shutdownGrace: cfg.ShutdownGrace,
+		close:         closers.Close,
 	}, nil
 }
 
@@ -323,21 +337,31 @@ func (a *devshardApp) Run(ctx context.Context) error {
 	}()
 
 	addr := fmt.Sprintf(":%d", a.port)
-	errCh := make(chan error, 1)
-	go func() {
-		slog.Info("listening", "addr", addr)
-		if err := a.server.Start(addr); err != nil && err != http.ErrServerClosed {
-			errCh <- err
-		}
-	}()
+	type serverError struct {
+		name string
+		err  error
+	}
+	errCh := make(chan serverError, 2)
+	startServer := func(name string, server *echo.Echo, addr string) {
+		go func() {
+			slog.Info("listening", "server", name, "addr", addr)
+			if err := server.Start(addr); err != nil && err != http.ErrServerClosed {
+				errCh <- serverError{name: name, err: err}
+			}
+		}()
+	}
+	startServer("public", a.server, addr)
+	if a.adminServer != nil {
+		startServer("admin", a.adminServer, a.adminAddr)
+	}
 
 	var runErr error
 	chainEventsStopped := false
 	select {
 	case <-ctx.Done():
 		slog.Info("shutdown requested")
 	case err := <-errCh:
-		runErr = fmt.Errorf("server error: %w", err)
+		runErr = fmt.Errorf("%s server error: %w", err.name, err.err)
 	case err := <-chainEventsErrCh:
 		chainEventsStopped = true
 		if err != nil {
@@ -347,11 +371,15 @@ func (a *devshardApp) Run(ctx context.Context) error {
 		}
 	}
 
+	a.lifecycle.StartDrain()
 	cancel()
 
-	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), a.shutdownGrace)
 	defer shutdownCancel()
 	_ = a.server.Shutdown(shutdownCtx)
+	if a.adminServer != nil {
+		_ = a.adminServer.Shutdown(shutdownCtx)
+	}
 	if !chainEventsStopped {
 		select {
 		case err := <-chainEventsErrCh:

devshard/cmd/devshardd/chain_events.go

@@ -80,6 +80,10 @@ func (b *chainEventBridge) OnNewBlock(h events.NewBlockHandler) {
 	b.listener.OnNewBlock(h)
 }
 
+func (b *chainEventBridge) OnReady(h func(bool)) {
+	b.listener.OnReady(h)
+}
+
 func (b *chainEventBridge) Start(ctx context.Context) error {
 	if err := b.listener.Start(ctx); err != nil && err != context.Canceled {
 		return err

devshard/cmd/devshardd/config.go

@@ -26,13 +26,15 @@ var sdkConfigOnce sync.Once
 
 type runtimeConfig struct {
 	Port                    int
+	AdminAddr               string
 	DataDir                 string
 	BinaryLogVersion        string
 	RuntimeVersion          string
 	ProtocolVersion         string
 	NodeManagerAddr         string
 	ValidationRetryInterval time.Duration
 	ValidationLeaseTTL      time.Duration
+	ShutdownGrace           time.Duration
 	Node                    ChainNodeConfig
 }
 
@@ -128,15 +130,22 @@ func loadRuntimeConfig(args []string, protocolVersion, linkBinaryVersion string)
 		return runtimeConfig{}, fmt.Errorf("DEVSHARD_VALIDATION_LEASE_TTL: %w", err)
 	}
 
+	shutdownGrace, err := parseDurationEnv("DEVSHARD_SHUTDOWN_GRACE", 10*time.Minute)
+	if err != nil {
+		return runtimeConfig{}, fmt.Errorf("DEVSHARD_SHUTDOWN_GRACE: %w", err)
+	}
+
 	return runtimeConfig{
 		Port:                    *port,
+		AdminAddr:               strings.TrimSpace(os.Getenv("DEVSHARD_ADMIN_ADDR")),
 		DataDir:                 *dataDir,
 		BinaryLogVersion:        binaryLogVersion,
 		RuntimeVersion:          protocolVersion,
 		ProtocolVersion:         protocolVersion,
 		NodeManagerAddr:         envOr("NODE_MANAGER_ADDR", "localhost:9400"),
 		ValidationRetryInterval: retryInterval,
 		ValidationLeaseTTL:      leaseTTL,
+		ShutdownGrace:           shutdownGrace,
 		Node:                    loadNodeConfigFromEnv(),
 	}, nil
 }

devshard/cmd/devshardd/events/listener.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"sync"
 	"time"
 
 	abci "github.com/cometbft/cometbft/abci/types"
@@ -67,6 +68,8 @@ type Listener struct {
 	rpcURL         string
 	subs           []subscription
 	reconnectDelay time.Duration
+	readyMu        sync.RWMutex
+	ready          func(bool)
 }
 
 // NewListener creates a Listener that connects to the given CometBFT RPC URL.
@@ -94,6 +97,23 @@ func (l *Listener) OnNewBlock(h NewBlockHandler) {
 	Subscribe(l, "tm.event='NewBlock'", parseNewBlockEvent, h)
 }
 
+// OnReady registers a callback invoked with true after all subscriptions are
+// active, and false when the listener disconnects or stops.
+func (l *Listener) OnReady(h func(bool)) {
+	l.readyMu.Lock()
+	l.ready = h
+	l.readyMu.Unlock()
+}
+
+func (l *Listener) setReady(ready bool) {
+	l.readyMu.RLock()
+	h := l.ready
+	l.readyMu.RUnlock()
+	if h != nil {
+		h(ready)
+	}
+}
+
 // Start connects to the chain and listens for events until ctx is cancelled.
 // Reconnects automatically on connection failure or subscription drop.
 func (l *Listener) Start(ctx context.Context) error {
@@ -147,6 +167,8 @@ func (l *Listener) run(ctx context.Context) error {
 			}
 		}()
 	}
+	l.setReady(true)
+	defer l.setReady(false)
 
 	select {
 	case <-ctx.Done():

devshard/cmd/devshardd/events/listener_internal_test.go

@@ -0,0 +1,20 @@
+package events
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestListenerReadyHandler(t *testing.T) {
+	l := NewListener("http://localhost:26657")
+	var states []bool
+	l.OnReady(func(ready bool) {
+		states = append(states, ready)
+	})
+
+	l.setReady(true)
+	l.setReady(false)
+
+	require.Equal(t, []bool{true, false}, states)
+}

devshard/cmd/devshardd/lifecycle.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"net/http"
+	"strings"
+	"sync/atomic"
+
+	"devshard/observability"
+
+	"github.com/labstack/echo/v4"
+)
+
+type lifecycleState struct {
+	ready    atomic.Bool
+	draining atomic.Bool
+	inflight atomic.Int64
+}
+
+type drainStatus struct {
+	Ready    bool  `json:"ready"`
+	Draining bool  `json:"draining"`
+	Inflight int64 `json:"inflight"`
+}
+
+func newLifecycleState() *lifecycleState {
+	observability.SetLifecycleInflight(0)
+	return &lifecycleState{}
+}
+
+func (s *lifecycleState) SetReady(ready bool) {
+	s.ready.Store(ready)
+}
+
+func (s *lifecycleState) StartDrain() {
+	s.draining.Store(true)
+	s.ready.Store(false)
+}
+
+func (s *lifecycleState) Status() drainStatus {
+	return drainStatus{
+		Ready:    s.ready.Load(),
+		Draining: s.draining.Load(),
+		Inflight: s.inflight.Load(),
+	}
+}
+
+func (s *lifecycleState) addInflight(delta int64) {
+	inflight := s.inflight.Add(delta)
+	observability.SetLifecycleInflight(inflight)
+}
+
+func (s *lifecycleState) middleware(next echo.HandlerFunc) echo.HandlerFunc {
+	return func(c echo.Context) error {
+		if isLifecycleBypassPath(c.Request().URL.Path) {
+			return next(c)
+		}
+		if s.draining.Load() {
+			return echo.NewHTTPError(http.StatusServiceUnavailable, "devshardd is draining")
+		}
+		s.addInflight(1)
+		defer s.addInflight(-1)
+		return next(c)
+	}
+}
+
+func isLifecycleBypassPath(path string) bool {
+	clean := "/" + strings.Trim(strings.TrimSpace(path), "/")
+	switch clean {
+	case "/healthz", "/metrics":
+		return true
+	default:
+		return false
+	}
+}