feat(ha): wire opt-in production runtime

server/cmd/fleetd/config.go

@@ -15,6 +15,7 @@ import (
 	"github.com/block/proto-fleet/server/internal/domain/telemetry/scheduler"
 	"github.com/block/proto-fleet/server/internal/domain/token"
 	"github.com/block/proto-fleet/server/internal/domain/updates"
+	"github.com/block/proto-fleet/server/internal/ha"
 	"github.com/block/proto-fleet/server/internal/infrastructure/db"
 	"github.com/block/proto-fleet/server/internal/infrastructure/encrypt"
 	"github.com/block/proto-fleet/server/internal/infrastructure/files"
@@ -60,6 +61,7 @@ type Config struct {
 	Files          files.Config                 `embed:"" prefix:"files-" envprefix:"FILES_"`
 	FleetTelemetry fleet_telemetry.Config       `embed:"" prefix:"fleet-telemetry-" envprefix:"FLEET_TELEMETRY_"`
 	Metrics        metrics.Config               `embed:"" prefix:"metrics-" envprefix:"FLEET_ALERTS_"`
+	HA             ha.Config                    `embed:"" prefix:"ha-" envprefix:"FLEET_HA_"`
 
 	SystemMonitoring sysmon.Config `embed:"" prefix:"system-monitoring-" envprefix:"FLEET_SYSTEM_MONITORING_"`
 }

server/cmd/fleetd/main.go

@@ -178,7 +178,15 @@ var reflectEnabledServices = []string{
 	instancev1connect.InstanceUpdateServiceName,
 }
 
-func start(config *Config) error {
+func start(config *Config) (result error) {
+	if err := config.HA.Validate(); err != nil {
+		return fmt.Errorf("invalid HA configuration: %w", err)
+	}
+	if config.HA.Enabled {
+		if err := config.DB.ValidateHA(); err != nil {
+			return fmt.Errorf("invalid HA database configuration: %w", err)
+		}
+	}
 	// Construct one configured registry before starting services. The CRUD
 	// service uses it now; the Phase 5 reconciler will share this same instance.
 	infrastructureDriverRegistry, err := infrastructureDomain.NewConfiguredDriverRegistry(config.Infrastructure)
@@ -668,13 +676,22 @@ func start(config *Config) error {
 	if err != nil {
 		return fmt.Errorf("create runtime job group: %w", err)
 	}
-	// HA configuration is not exposed yet, so production stays standalone.
-	fleetRuntime, err := ha.NewStandaloneRuntime(runtimeJobGroup, executionService.IsRunning)
+	fleetRuntime, closeHA, err := ha.NewConfiguredRuntime(
+		config.HA,
+		conn,
+		runtimeJobGroup,
+		executionService.IsRunning,
+	)
 	if err != nil {
 		return fmt.Errorf("create Fleet runtime: %w", err)
 	}
 	defer func() {
-		stopRuntimeJobGroup(runtimeJobGroup, executionService, shutdownTimeout)
+		if err := closeHA(); err != nil {
+			slog.Error("Failed to close HA services", "error", err)
+		}
+	}()
+	defer func() {
+		stopRuntimeJobGroupAfterRun(result, runtimeJobGroup, executionService, shutdownTimeout)
 	}()
 
 	middlewares := []server.Middleware{

server/cmd/fleetd/main_test.go

@@ -121,6 +121,33 @@ encrypt:
 	require.Equal(t, explicitDSN, config.DB.ExplicitDSN)
 }
 
+func TestFleetdParsesHAEnabledFromEnv(t *testing.T) {
+	t.Setenv("FLEET_HA_ENABLED", "true")
+	t.Setenv("FLEET_HA_ETCD_ENDPOINTS", "https://10.0.0.1:2379,https://10.0.0.2:2379")
+
+	configPath := writeFleetdConfigFile(t, `
+auth:
+  client:
+    expiration-period: "1h"
+    secret-key: "test-client-secret"
+  miner-token-expiration-period: "30m"
+encrypt:
+  service-master-key: "test-master-key"
+`)
+	config := &Config{}
+	parser, err := kong.New(
+		config,
+		kong.Name("fleetd"),
+		kong.Configuration(kongyaml.Loader, configPath),
+	)
+	require.NoError(t, err)
+	_, err = parser.Parse(nil)
+	require.NoError(t, err)
+	require.True(t, config.HA.Enabled)
+	require.Equal(t, []string{"https://10.0.0.1:2379", "https://10.0.0.2:2379"}, config.HA.EtcdEndpoints)
+	require.NoError(t, config.HA.Validate())
+}
+
 func TestFleetdInfrastructureOTControlSubnetsFlag(t *testing.T) {
 	t.Parallel()
 

server/cmd/fleetd/runtime_jobs.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/block/proto-fleet/server/internal/ha"
 	"github.com/block/proto-fleet/server/internal/runtimejobs"
 )
 
@@ -169,6 +170,19 @@ func serveFleetRuntime(
 	}
 }
 
+// stopRuntimeJobGroupAfterRun avoids repeating cleanup already completed by a fatal HA abort.
+func stopRuntimeJobGroupAfterRun(
+	runErr error,
+	group runtimeJobGroupStopper,
+	commandExecution runtimejobs.Lifecycle,
+	timeout time.Duration,
+) {
+	if errors.Is(runErr, ha.ErrRuntimeAborted) {
+		return
+	}
+	stopRuntimeJobGroup(group, commandExecution, timeout)
+}
+
 // stopRuntimeJobGroup gives the group one graceful-shutdown budget. Command
 // execution receives a final independent budget because its activation is
 // detached from group cancellation to preserve shutdown ordering.

server/cmd/fleetd/runtime_jobs_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/block/proto-fleet/server/internal/ha"
 	"github.com/block/proto-fleet/server/internal/runtimejobs"
 	"github.com/stretchr/testify/require"
 )
@@ -233,6 +234,39 @@ func TestStopRuntimeJobGroupDoesNotRetrySuccessfulStop(t *testing.T) {
 	require.True(t, hasDeadline)
 }
 
+func TestStopRuntimeJobGroupAfterRunOnlySkipsHAAbort(t *testing.T) {
+	t.Run("HA abort", func(t *testing.T) {
+		// Arrange
+		group := &scriptedRuntimeJobGroupStopper{
+			stop: func(context.Context) error { return nil },
+		}
+
+		// Act
+		stopRuntimeJobGroupAfterRun(
+			errors.Join(errors.New("runtime failed"), ha.ErrRuntimeAborted),
+			group,
+			noopLifecycle{},
+			time.Second,
+		)
+
+		// Assert
+		require.Empty(t, group.contexts)
+	})
+
+	t.Run("standalone failure", func(t *testing.T) {
+		// Arrange
+		group := &scriptedRuntimeJobGroupStopper{
+			stop: func(context.Context) error { return nil },
+		}
+
+		// Act
+		stopRuntimeJobGroupAfterRun(errors.New("runtime failed"), group, noopLifecycle{}, time.Second)
+
+		// Assert
+		require.Len(t, group.contexts, 1)
+	})
+}
+
 func TestStopRuntimeJobGroupStopsCommandAfterGroupFailure(t *testing.T) {
 	group := &scriptedRuntimeJobGroupStopper{
 		stop: func(context.Context) error { return errors.New("stop failed") },

server/internal/domain/curtailment/reconciler/reconciler.go

@@ -141,6 +141,7 @@ type Reconciler struct {
 }
 
 var _ runtimejobs.Lifecycle = (*Reconciler)(nil)
+var _ runtimejobs.Aborter = (*Reconciler)(nil)
 
 // Option configures a Reconciler at construction time.
 type Option func(*Reconciler)
@@ -280,6 +281,20 @@ func (r *Reconciler) Stop(ctx context.Context) error {
 	}
 }
 
+// Abort immediately cancels admission and detached work before a fatal exit.
+func (r *Reconciler) Abort() {
+	r.mu.Lock()
+	loopCancel := r.loopCancel
+	workCancel := r.workCancel
+	r.mu.Unlock()
+	if loopCancel != nil {
+		loopCancel()
+	}
+	if workCancel != nil {
+		workCancel()
+	}
+}
+
 func (r *Reconciler) tickLoop(loopCtx, workCtx context.Context, runDone chan<- struct{}) {
 	defer close(runDone)
 	defer r.finishActivation()

server/internal/domain/curtailment/reconciler/reconciler_test.go

@@ -4426,6 +4426,49 @@ func TestReconciler_StopDeadlineCancelsInFlightWork(t *testing.T) {
 	require.NoError(t, r.Stop(context.Background()))
 }
 
+func TestReconciler_AbortCancelsDetachedWork(t *testing.T) {
+	// Arrange
+	store := newFakeStore()
+	workStarted := make(chan struct{})
+	workCanceled := make(chan struct{})
+	store.listEventsHook = func(ctx context.Context) {
+		close(workStarted)
+		<-ctx.Done()
+		close(workCanceled)
+	}
+	r := New(Config{TickInterval: time.Hour}, store, &fakeDispatcher{})
+	loopCtx, loopCancel := context.WithCancel(t.Context())
+	workCtx, workCancel := context.WithCancel(context.WithoutCancel(t.Context()))
+	r.loopCancel = loopCancel
+	r.workCancel = workCancel
+	workDone := make(chan struct{})
+	go func() {
+		defer close(workDone)
+		r.safeTick(workCtx)
+	}()
+	<-workStarted
+
+	// Act
+	r.Abort()
+
+	// Assert
+	select {
+	case <-loopCtx.Done():
+	case <-time.After(time.Second):
+		t.Fatal("Abort did not close curtailment admission")
+	}
+	select {
+	case <-workCanceled:
+	case <-time.After(time.Second):
+		t.Fatal("Abort did not cancel detached curtailment work")
+	}
+	select {
+	case <-workDone:
+	case <-time.After(time.Second):
+		t.Fatal("canceled curtailment work did not return")
+	}
+}
+
 func TestReconciler_StopCancellationPreventsOverlappingRestart(t *testing.T) {
 	store := newFakeStore()
 	disp := &fakeDispatcher{}

server/internal/domain/schedule/processor.go

@@ -110,6 +110,7 @@ type Processor struct {
 }
 
 var _ runtimejobs.Lifecycle = (*Processor)(nil)
+var _ runtimejobs.Aborter = (*Processor)(nil)
 
 func NewProcessor(
 	procStore interfaces.ScheduleProcessorStore,
@@ -229,6 +230,18 @@ func (p *Processor) Stop(ctx context.Context) error {
 	}
 }
 
+// Abort immediately cancels admission and detached work before a fatal exit.
+func (p *Processor) Abort() {
+	p.lifecycleMu.Lock()
+	run := p.activation
+	p.lifecycleMu.Unlock()
+	if run == nil {
+		return
+	}
+	p.beginStop(run)
+	run.cancelWork()
+}
+
 func (p *Processor) beginStop(run *processorActivation) {
 	run.stopOnce.Do(func() {
 		run.cancelAdmission()

server/internal/domain/schedule/processor_test.go

@@ -296,6 +296,31 @@ func TestProcessor_StopDeadlineForceCancelsAdmittedWork(t *testing.T) {
 	assert.NoError(t, p.Stop(context.Background()))
 }
 
+func TestProcessor_AbortCancelsAdmittedWork(t *testing.T) {
+	// Arrange
+	run := newProcessorActivation(t.Context())
+	close(run.startupDone)
+	p := &Processor{
+		activation: run,
+		jobs:       make(map[int64]jobEntry),
+	}
+	workCanceled := make(chan struct{})
+	run.wg.Add(1)
+	go func() {
+		defer run.wg.Done()
+		<-run.workCtx.Done()
+		close(workCanceled)
+	}()
+
+	// Act
+	p.Abort()
+
+	// Assert
+	waitForSignal(t, run.admissionCtx.Done(), "Abort did not close schedule admission")
+	waitForSignal(t, workCanceled, "Abort did not cancel admitted schedule work")
+	assert.NoError(t, p.Stop(t.Context()))
+}
+
 func TestProcessor_ActivationCancellationPreventsNewTimerWork(t *testing.T) {
 	p, _, _, _, _ := newTestProcessor(t, time.Now())
 	run := newProcessorActivation(t.Context())