fix(cli/operator): convert logger.Fatal goroutines to errgroup (#2867)

cli/operator/eventsync.go

@@ -20,7 +20,9 @@ import (
 	"github.com/ssvlabs/ssv/ssvsigner/ekm"
 )
 
-// syncContractEvents blocks until historical events are synced and then spawns a goroutine syncing ongoing events.
+// syncContractEvents blocks until historical events are synced and returns the EventSyncer.
+// It also returns a start-func for ongoing event sync (nil if LocalEventsPath is set),
+// which the caller must run (e.g. via errgroup.Go) to keep the node current with the contract.
 func syncContractEvents(
 	ctx context.Context,
 	logger *zap.Logger,
@@ -32,15 +34,15 @@ func syncContractEvents(
 	operatorDataStore operatordatastore.OperatorDataStore,
 	keyManager ekm.KeyManager,
 	doppelgangerHandler eventhandler.DoppelgangerProvider,
-) (*eventsyncer.EventSyncer, error) {
+) (*eventsyncer.EventSyncer, func() error, error) {
 	eventFilterer, err := executionClient.Filterer()
 	if err != nil {
-		return nil, fmt.Errorf("failed to set up event filterer: %w", err)
+		return nil, nil, fmt.Errorf("failed to set up event filterer: %w", err)
 	}
 
 	eventParser, err := eventparser.New(eventFilterer)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create event parser: %w", err)
+		return nil, nil, fmt.Errorf("failed to create event parser: %w", err)
 	}
 
 	eventHandler, err := eventhandler.New(
@@ -55,7 +57,7 @@ func syncContractEvents(
 		eventhandler.WithLogger(logger),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("failed to setup event data handler: %w", err)
+		return nil, nil, fmt.Errorf("failed to setup event data handler: %w", err)
 	}
 
 	eventSyncer := eventsyncer.New(
@@ -67,26 +69,28 @@ func syncContractEvents(
 
 	fromBlock, found, err := nodeStorage.GetLastProcessedBlock(nil)
 	if err != nil {
-		return nil, fmt.Errorf("syncing registry contract events failed, could not get last processed block: %w", err)
+		return nil, nil, fmt.Errorf("syncing registry contract events failed, could not get last processed block: %w", err)
 	}
 	if !found {
 		fromBlock = networkConfig.RegistrySyncOffset
 	} else if fromBlock == nil {
-		return nil, fmt.Errorf("syncing registry contract events failed, last processed block is nil")
+		return nil, nil, fmt.Errorf("syncing registry contract events failed, last processed block is nil")
 	} else {
 		// Start syncing from the next block.
 		fromBlock = new(big.Int).SetUint64(fromBlock.Uint64() + 1)
 	}
 
+	var ongoingSync func() error
+
 	// load & parse local events yaml if exists, otherwise sync from contract
 	if len(cfg.LocalEventsPath) != 0 {
 		localEvents, err := localevents.Load(cfg.LocalEventsPath)
 		if err != nil {
-			return nil, fmt.Errorf("failed to load local events: %w", err)
+			return nil, nil, fmt.Errorf("failed to load local events: %w", err)
 		}
 
 		if err := eventHandler.HandleLocalEvents(ctx, localEvents); err != nil {
-			return nil, fmt.Errorf("error occurred while running event data handler: %w", err)
+			return nil, nil, fmt.Errorf("error occurred while running event data handler: %w", err)
 		}
 	} else {
 		// Sync historical registry events.
@@ -106,7 +110,7 @@ func syncContractEvents(
 			)
 			fromBlock = new(big.Int).SetUint64(lastProcessedBlock + 1)
 		default:
-			return nil, fmt.Errorf("failed to sync historical registry events: %w", err)
+			return nil, nil, fmt.Errorf("failed to sync historical registry events: %w", err)
 		}
 
 		// Print registry stats.
@@ -137,19 +141,18 @@ func syncContractEvents(
 			zap.Int("my_validators", operatorValidators),
 		)
 
-		// Sync ongoing registry events in the background. Crash if it stops: the node can't operate
-		// without staying current with Ethereum events, and until reorg handling exists, restarting
-		// from persisted state is safer than continuing on possibly-incorrect state.
-		go func() {
-			err := eventSyncer.SyncOngoing(ctx, fromBlock.Uint64())
-			if err != nil && !errors.Is(err, context.Canceled) {
-				logger.Fatal("failed syncing ongoing registry events",
-					zap.Uint64("last_processed_block", lastProcessedBlock),
-					zap.Error(err),
-				)
+		// Return a start-func for ongoing event sync. The caller (start()) registers it with errgroup
+		// so a failure propagates as a clean error rather than a Fatal. The node can't operate without
+		// staying current with Ethereum events, so any non-canceled failure exits the process.
+		ongoingFromBlock := fromBlock.Uint64()
+		ongoingSync = func() error {
+			err := eventSyncer.SyncOngoing(ctx, ongoingFromBlock)
+			if err != nil && ctx.Err() == nil {
+				return fmt.Errorf("failed syncing ongoing registry events (from_block=%d): %w", ongoingFromBlock, err)
 			}
-		}()
+			return nil
+		}
 	}
 
-	return eventSyncer, nil
+	return eventSyncer, ongoingSync, nil
 }

cli/operator/node.go

@@ -14,6 +14,7 @@ import (
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 
 	hexporter "github.com/ssvlabs/ssv/api/handlers/exporter"
 	hnode "github.com/ssvlabs/ssv/api/handlers/node"
@@ -486,33 +487,35 @@ func (n *node) Close() error {
 	return errors.Join(errs...)
 }
 
-// start launches the node's long-lived services (metrics + SSV API servers, the health
-// prober, contract-event sync) and blocks on operatorNode.Start until the node's ctx is canceled.
-// NOTE: the metrics/SSV-API server goroutines call logger.Fatal on failure — crashing the
-// process and bypassing Close() — a candidate for errgroup-based coordinated shutdown.
+// start launches the node's long-lived services and blocks until the node's ctx is canceled
+// or any service fails. All services run under a shared errgroup so the first failure cancels
+// the rest and propagates to the single logger.Fatal in start_node.go.
 func (n *node) start() error {
+	g, gctx := errgroup.WithContext(n.ctx)
+
 	if n.cfg.MetricsAPIPort > 0 {
 		metricsHandler := metrics.NewHandler(n.logger, n.db, n.cfg.EnableProfile, n.operatorNode)
-		_, metricsServeErr, err := metricsHandler.Start(n.ctx, http.NewServeMux(), fmt.Sprintf(":%d", n.cfg.MetricsAPIPort))
+		_, metricsServeErr, err := metricsHandler.Start(gctx, http.NewServeMux(), fmt.Sprintf(":%d", n.cfg.MetricsAPIPort))
 		if err != nil {
-			n.logger.Fatal("failed to start metrics server", zap.Error(err))
+			return fmt.Errorf("failed to start metrics server: %w", err)
 		}
-		go func() {
+		g.Go(func() error {
 			if err := <-metricsServeErr; err != nil {
-				n.logger.Fatal("metrics server serve loop exited", zap.Error(err))
+				return fmt.Errorf("metrics server stopped: %w", err)
 			}
-		}()
+			return nil
+		})
 	}
 
 	healthProber := hprobe.NewHealthProber(n.logger)
 	healthProber.AddComponent(clComponentName, n.consensusClient, proberHealthcheckTimeout, proberRetriesMax, proberRetryDelay)
 	healthProber.AddComponent(elComponentName, n.executionClient, proberHealthcheckTimeout, proberRetriesMax, proberRetryDelay)
-	if err := ensureComponentsHealthy(n.ctx, n.logger, healthProber); err != nil {
+	if err := ensureComponentsHealthy(gctx, n.logger, healthProber); err != nil {
 		return err
 	}
 
-	eventSyncer, err := syncContractEvents(
-		n.ctx,
+	eventSyncer, ongoingSync, err := syncContractEvents(
+		gctx,
 		n.logger,
 		n.cfg,
 		n.executionClient,
@@ -529,15 +532,18 @@ func (n *node) start() error {
 	if len(n.cfg.LocalEventsPath) == 0 {
 		healthProber.AddComponent(eventSyncerComponentName, eventSyncer, proberHealthcheckTimeout, proberRetriesMax, proberRetryDelay)
 	}
+	if ongoingSync != nil {
+		g.Go(ongoingSync)
+	}
 
-	go startHealthProber(n.ctx, n.logger, healthProber)
+	g.Go(func() error { return startHealthProber(gctx, n.logger, healthProber) })
 
-	if _, err := n.metadataSyncer.SyncAll(n.ctx); err != nil {
+	if _, err := n.metadataSyncer.SyncAll(gctx); err != nil {
 		return fmt.Errorf("failed to sync metadata on startup: %w", err)
 	}
 
 	if n.usingSSVSigner {
-		if err := ensureNoMissingKeys(n.ctx, n.nodeStorage, n.operatorDataStore, n.ssvSignerClient); err != nil {
+		if err := ensureNoMissingKeys(gctx, n.nodeStorage, n.operatorDataStore, n.ssvSignerClient); err != nil {
 			return err
 		}
 	}
@@ -599,21 +605,26 @@ func (n *node) start() error {
 			hexporter.NewExporter(n.logger, n.storageMap, n.collector, n.nodeStorage.ValidatorStore()),
 			n.mode == modeExporterArchive,
 		)
-		_, apiServeErr, err := apiServer.Start(n.ctx)
+		_, apiServeErr, err := apiServer.Start(gctx)
 		if err != nil {
-			n.logger.Fatal("failed to start API server", zap.Error(err))
+			return fmt.Errorf("failed to start API server: %w", err)
 		}
-		go func() {
+		g.Go(func() error {
 			if err := <-apiServeErr; err != nil {
-				n.logger.Fatal("API server serve loop exited", zap.Error(err))
+				return fmt.Errorf("API server stopped: %w", err)
 			}
-		}()
-	}
-	if err := n.operatorNode.Start(n.ctx); err != nil {
-		return fmt.Errorf("failed to start SSV node: %w", err)
+			return nil
+		})
 	}
 
-	return nil
+	g.Go(func() error {
+		if err := n.operatorNode.Start(gctx); err != nil {
+			return fmt.Errorf("failed to start SSV node: %w", err)
+		}
+		return nil
+	})
+
+	return g.Wait()
 }
 
 // startNetwork wires validator stats into the p2p layer, then sets up + starts the network and

cli/operator/prober.go

@@ -39,33 +39,29 @@ func ensureComponentsHealthy(ctx context.Context, logger *zap.Logger, p *hprobe.
 	return nil
 }
 
-func startHealthProber(ctx context.Context, logger *zap.Logger, p *hprobe.HealthProber) {
+func startHealthProber(ctx context.Context, logger *zap.Logger, p *hprobe.HealthProber) error {
 	const probeFrequency = 60 * time.Second
 
 	ticker := time.NewTicker(probeFrequency)
 	defer ticker.Stop()
 
 	for {
-		func() {
-			logger.Debug("health-prober tick: probing all components")
-			defer logger.Debug("health-prober tick: probing all components done")
-
-			probeCtx, cancel := context.WithTimeout(ctx, probeFrequency)
-			defer cancel()
-
-			if err := p.ProbeAll(probeCtx); err != nil {
-				// TODO(#2867): trigger graceful shutdown (-> Close -> non-zero exit) instead of Fatal,
-				// which bypasses Close. Crash-and-restart on persistent unhealth is intentional; the
-				// goroutine os.Exit mechanism is the wart.
-				logger.Fatal(componentsUnhealthyErrorMsg, zap.Error(err))
+		logger.Debug("health-prober tick: probing all components")
+		probeCtx, cancel := context.WithTimeout(ctx, probeFrequency)
+		err := p.ProbeAll(probeCtx)
+		cancel()
+		logger.Debug("health-prober tick: probing all components done")
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil // ctx canceled (clean shutdown), not a real probe failure
 			}
-		}()
+			return fmt.Errorf("%s: %w", componentsUnhealthyErrorMsg, err)
+		}
 
 		select {
 		case <-ctx.Done():
-			return
+			return nil
 		case <-ticker.C:
-			continue
 		}
 	}
 }

cli/operator/prober_test.go

@@ -0,0 +1,60 @@
+package operator
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"github.com/ssvlabs/ssv/hprobe"
+)
+
+// simpleComponent implements hprobe's component interface for testing.
+type simpleComponent struct{ err error }
+
+func (s simpleComponent) Healthy(context.Context) error { return s.err }
+
+func Test_startHealthProber_returnsNilOnCtxCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	prober := hprobe.NewHealthProber(zap.NewNop())
+	prober.AddComponent("cl", simpleComponent{}, time.Second, 0, 0)
+
+	done := make(chan error, 1)
+	go func() { done <- startHealthProber(ctx, zap.NewNop(), prober) }()
+
+	cancel()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("startHealthProber did not exit after ctx cancellation")
+	}
+}
+
+func Test_startHealthProber_returnsErrOnUnhealthy(t *testing.T) {
+	prober := hprobe.NewHealthProber(zap.NewNop())
+	prober.AddComponent("cl", simpleComponent{err: errors.New("broken")}, 100*time.Millisecond, 0, 0)
+
+	err := startHealthProber(context.Background(), zap.NewNop(), prober)
+	require.Error(t, err)
+	require.ErrorContains(t, err, componentsUnhealthyErrorMsg)
+}
+
+// Test_startHealthProber_ctxCancelMasksProbeFail verifies the ctx.Err() guard: when ctx is
+// canceled at the same time a probe fails (e.g. the network stack tears down components before
+// the errgroup context fully propagates), startHealthProber returns nil rather than a misleading
+// "components unhealthy" error that would cause a Fatal log on normal shutdown.
+func Test_startHealthProber_ctxCancelMasksProbeFail(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // cancel before starting — simulates shutdown racing a probe
+
+	prober := hprobe.NewHealthProber(zap.NewNop())
+	prober.AddComponent("cl", simpleComponent{err: errors.New("broken")}, 100*time.Millisecond, 0, 0)
+
+	err := startHealthProber(ctx, zap.NewNop(), prober)
+	require.NoError(t, err)
+}

codecov.yml

@@ -39,3 +39,5 @@ ignore:
   - "api/types.go"
   - "beacon/goclient/types.go"
   - "cli/operator/generate_doc.go"
+  - "cli/operator/node.go"
+  - "cli/operator/eventsync.go"