Complete HA updates through bounded failover

deployment-files/ha/README.md

@@ -164,6 +164,43 @@ The Docker repository setup follows the official instructions for
 [Ubuntu](https://docs.docker.com/engine/install/ubuntu/), and
 [64-bit Raspberry Pi OS](https://docs.docker.com/engine/install/raspberry-pi-os/).
 
+## Update a passive Fleet host
+
+HA disables application-triggered updates. On the passive database host, run:
+
+```bash
+sudo /opt/proto-fleet/deployment/ha/fleet-ha update v0.2.11
+```
+
+The local updater downloads the release from the fixed Proto Fleet GitHub
+release origin, verifies its SHA256 checksum, builds and persists the staged
+Fleet images, then rechecks that this host is passive. It stops and replaces
+only `fleet-api` and `fleet-client`; etcd, Patroni, PostgreSQL, and keepalived
+remain running. The command returns only after the target version is healthy
+and passive.
+
+After the peer is confirmed on the target release, complete the update from
+the old active host:
+
+```bash
+sudo /opt/proto-fleet/deployment/ha/fleet-ha update v0.2.11 --complete
+```
+
+If the updated host has already become active, the old host is now passive.
+Run the ordinary `fleet-ha update v0.2.11` command on that passive host instead.
+
+The source release must already contain this HA update protocol. HA is being
+introduced for new deployments, so an older experimental HA installation that
+predates `update --complete` must be reinstalled at the supported baseline
+rather than upgraded through this workflow.
+
+The updater stages everything first, stops the local Fleet containers, and
+waits for the updated peer to serve the VIP with the target version. Only then
+does it swap and restart the local application as passive. If takeover does
+not complete within 35 seconds, it restarts the old local release without
+swapping. Qualification still requires a healthy takeover in less than 15
+seconds. This is a bounded interruption, not a zero-downtime update.
+
 ## Qualification
 
 The distributions above are installer-compatible targets. The HA profile is

server/cmd/fleet-ha/main.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/google/uuid"
 
+	"github.com/block/proto-fleet/server/internal/ha"
 	"github.com/block/proto-fleet/server/internal/ha/deployment"
 	"github.com/block/proto-fleet/server/internal/updaterapi"
 )
@@ -91,7 +92,7 @@ func run(ctx context.Context, args []string) error {
 	case "install":
 		return runInstall(ctx, args[1:])
 	case "update":
-		return runPassiveUpdate(ctx, args[1:], os.Stdout, deployment.ValidatePassiveUpdate, updaterapi.NewClient(defaultUpdaterSocket), deployment.Status)
+		return runPassiveUpdate(ctx, args[1:], os.Stdout, validateHAUpdate, updaterapi.NewClient(defaultUpdaterSocket), deployment.Status)
 	case "start":
 		return runStart(ctx, args[1:])
 	case "stop":
@@ -104,6 +105,12 @@ func run(ctx context.Context, args []string) error {
 			return errors.New("usage: fleet-ha require-passive NODE_ENV VERSION")
 		}
 		return deployment.ValidatePassiveUpdate(ctx, args[1], args[2])
+	case "require-active":
+		if len(args) != 3 {
+			return errors.New("usage: fleet-ha require-active NODE_ENV VERSION")
+		}
+		_, err := deployment.ValidateActiveUpdate(ctx, args[1], args[2])
+		return err
 	case "update-preflight":
 		if len(args) != 1 {
 			return errors.New("usage: fleet-ha update-preflight")
@@ -114,24 +121,30 @@ func run(ctx context.Context, args []string) error {
 		}
 		return deployment.PrepareApplicationUpdate(ctx, root)
 	case "app-stop":
-		if len(args) != 1 {
-			return errors.New("usage: fleet-ha app-stop")
+		if len(args) != 2 || (args[1] != "passive" && args[1] != "active") {
+			return errors.New("usage: fleet-ha app-stop <passive|active>")
 		}
 		root, err := deployment.ReleaseRoot()
 		if err != nil {
 			return err
 		}
-		return deployment.StopApplication(ctx, root)
+		return deployment.StopApplication(ctx, root, ha.RuntimeRole(args[1]))
 	case "app-start":
-		if len(args) != 2 && (len(args) != 3 || (args[2] != "passive" && args[2] != "any")) {
-			return errors.New("usage: fleet-ha app-start VERSION <passive|any>")
+		if len(args) != 2 && (len(args) != 3 || (args[2] != "passive" && args[2] != "complete" && args[2] != "any")) {
+			return errors.New("usage: fleet-ha app-start VERSION <passive|complete|any>")
 		}
 		root, err := deployment.ReleaseRoot()
 		if err != nil {
 			return err
 		}
-		requirePassive := len(args) == 2 || args[2] == "passive"
-		return deployment.StartApplication(ctx, root, args[1], requirePassive)
+		requirePassive := len(args) == 2 || args[2] != "any"
+		requireFailoverReady := len(args) == 3 && args[2] == "complete"
+		return deployment.StartApplication(ctx, root, args[1], requirePassive, requireFailoverReady)
+	case "wait-takeover":
+		if len(args) != 2 {
+			return errors.New("usage: fleet-ha wait-takeover VERSION")
+		}
+		return deployment.WaitForVIPVersion(ctx, installedNodeEnv, args[1])
 	default:
 		return usageError()
 	}
@@ -149,6 +162,7 @@ const (
 type updaterClient interface {
 	Status(ctx context.Context) (updaterapi.StatusResponse, error)
 	Trigger(ctx context.Context, operationID, targetVersion string) (updaterapi.Operation, error)
+	TriggerComplete(ctx context.Context, operationID, targetVersion string) (updaterapi.Operation, error)
 }
 
 const (
@@ -158,7 +172,15 @@ const (
 
 type updateTrigger func(context.Context, string, string) (updaterapi.Operation, error)
 
-type updatePreflight func(context.Context, string, string) error
+type updatePreflight func(context.Context, string, string, bool) error
+
+func validateHAUpdate(ctx context.Context, envPath, targetVersion string, complete bool) error {
+	if complete {
+		_, err := deployment.ValidateActiveUpdate(ctx, envPath, targetVersion)
+		return err
+	}
+	return deployment.ValidatePassiveUpdate(ctx, envPath, targetVersion)
+}
 
 func runPassiveUpdate(
 	ctx context.Context,
@@ -168,6 +190,7 @@ func runPassiveUpdate(
 	client updaterClient,
 	read statusReader,
 ) error {
+	complete := len(args) == 2 && args[1] == "--complete"
 	if err := runUpdate(ctx, args, output, preflight, client); err != nil {
 		return err
 	}
@@ -178,7 +201,7 @@ func runPassiveUpdate(
 	if report.Control != nil && report.Control.FailoverReady {
 		return nil
 	}
-	if deployment.ExpectedRollingVersionMismatch(report.Control) {
+	if !complete && deployment.ExpectedRollingVersionMismatch(report.Control) {
 		_, err = fmt.Fprintln(output, "Update succeeded; failover readiness will recover after the peer is updated.")
 		if err != nil {
 			return fmt.Errorf("write update outcome: %w", err)
@@ -198,14 +221,19 @@ func runUpdate(
 	preflight updatePreflight,
 	client updaterClient,
 ) error {
-	if len(args) != 1 {
-		return errors.New("usage: fleet-ha update VERSION")
+	complete := len(args) == 2 && args[1] == "--complete"
+	if len(args) != 1 && !complete {
+		return errors.New("usage: fleet-ha update VERSION [--complete]")
 	}
-	if err := preflight(ctx, installedNodeEnv, args[0]); err != nil {
+	if err := preflight(ctx, installedNodeEnv, args[0], complete); err != nil {
 		return err
 	}
 	operationID := uuid.NewString()
-	operation, err := triggerUpdate(ctx, operationID, args[0], client.Trigger)
+	trigger := updateTrigger(client.Trigger)
+	if complete {
+		trigger = client.TriggerComplete
+	}
+	operation, err := triggerUpdate(ctx, operationID, args[0], trigger)
 	if err != nil {
 		return err
 	}

server/cmd/fleet-ha/main_test.go

@@ -15,10 +15,20 @@ import (
 
 type fakeUpdaterClient struct {
 	triggered  bool
+	complete   bool
 	triggerErr error
 	operation  updaterapi.Operation
 }
 
+func (f *fakeUpdaterClient) TriggerComplete(_ context.Context, operationID, targetVersion string) (updaterapi.Operation, error) {
+	f.triggered = true
+	f.complete = true
+	if f.triggerErr != nil {
+		return updaterapi.Operation{}, f.triggerErr
+	}
+	return updaterapi.Operation{ID: operationID, TargetVersion: targetVersion, Phase: updaterapi.PhaseSucceeded}, nil
+}
+
 func (f *fakeUpdaterClient) Status(context.Context) (updaterapi.StatusResponse, error) {
 	return updaterapi.StatusResponse{}, nil
 }
@@ -79,7 +89,7 @@ func TestUpdateRequiresPassiveBeforeTriggering(t *testing.T) {
 	client := &fakeUpdaterClient{}
 
 	// Act
-	err := runUpdate(t.Context(), []string{"v1.2.3"}, &bytes.Buffer{}, func(context.Context, string, string) error {
+	err := runUpdate(t.Context(), []string{"v1.2.3"}, &bytes.Buffer{}, func(context.Context, string, string, bool) error {
 		return errors.New("local Fleet is active")
 	}, client)
 
@@ -98,7 +108,7 @@ func TestUpdateReportsTerminalSuccess(t *testing.T) {
 	var output bytes.Buffer
 
 	// Act
-	err := runUpdate(t.Context(), []string{"v1.2.3"}, &output, func(context.Context, string, string) error { return nil }, client)
+	err := runUpdate(t.Context(), []string{"v1.2.3"}, &output, func(context.Context, string, string, bool) error { return nil }, client)
 
 	// Assert
 	require.NoError(t, err)
@@ -120,7 +130,7 @@ func TestPassiveUpdateReportsDegradedFailoverReadiness(t *testing.T) {
 	}
 
 	// Act
-	err := runPassiveUpdate(t.Context(), []string{"v1.2.3"}, &output, func(context.Context, string, string) error { return nil }, client, read)
+	err := runPassiveUpdate(t.Context(), []string{"v1.2.3"}, &output, func(context.Context, string, string, bool) error { return nil }, client, read)
 
 	// Assert
 	require.ErrorContains(t, err, "failover readiness is degraded")
@@ -138,18 +148,35 @@ func TestPassiveUpdateAllowsExpectedVersionMismatch(t *testing.T) {
 	}
 
 	// Act
-	err := runPassiveUpdate(t.Context(), []string{"v1.2.3"}, &bytes.Buffer{}, func(context.Context, string, string) error { return nil }, client, read)
+	err := runPassiveUpdate(t.Context(), []string{"v1.2.3"}, &bytes.Buffer{}, func(context.Context, string, string, bool) error { return nil }, client, read)
 
 	// Assert
 	require.NoError(t, err)
 }
 
+func TestCompleteUpdateRejectsExpectedVersionMismatch(t *testing.T) {
+	// Arrange
+	client := &fakeUpdaterClient{}
+	read := func(context.Context, string) (deployment.StatusReport, error) {
+		return deployment.StatusReport{Control: &deployment.ControlStatus{
+			ControlReady: true,
+			ReasonCodes:  []deployment.ControlReasonCode{deployment.ReasonFleetVersionMismatch},
+		}}, nil
+	}
+
+	// Act
+	err := runPassiveUpdate(t.Context(), []string{"v1.2.3", "--complete"}, &bytes.Buffer{}, func(context.Context, string, string, bool) error { return nil }, client, read)
+
+	// Assert
+	require.ErrorContains(t, err, "failover readiness is degraded")
+}
+
 func TestUpdateReturnsWhenUpdaterIsUnavailable(t *testing.T) {
 	// Arrange
 	client := &fakeUpdaterClient{triggerErr: updaterapi.ErrUnavailable}
 
 	// Act
-	err := runUpdate(t.Context(), []string{"v1.2.3"}, &bytes.Buffer{}, func(context.Context, string, string) error { return nil }, client)
+	err := runUpdate(t.Context(), []string{"v1.2.3"}, &bytes.Buffer{}, func(context.Context, string, string, bool) error { return nil }, client)
 
 	// Assert
 	require.ErrorIs(t, err, updaterapi.ErrUnavailable)
@@ -164,10 +191,39 @@ func TestUpdateFailureIncludesRecoveryDetails(t *testing.T) {
 	var output bytes.Buffer
 
 	// Act
-	err := runUpdate(t.Context(), []string{"v1.2.3"}, &output, func(context.Context, string, string) error { return nil }, client)
+	err := runUpdate(t.Context(), []string{"v1.2.3"}, &output, func(context.Context, string, string, bool) error { return nil }, client)
 
 	// Assert
 	require.ErrorContains(t, err, "Recovery: fleet-ha app-start v1.2.3")
 	require.ErrorContains(t, err, "Log: /var/log/proto-fleet-updater/update.log")
 	require.Contains(t, output.String(), "Update operation")
 }
+
+func TestCompleteUpdateRequiresActiveAndUsesCompletionRequest(t *testing.T) {
+	// Arrange
+	client := &fakeUpdaterClient{}
+	activeChecked := false
+
+	// Act
+	err := runUpdate(
+		t.Context(), []string{"v1.2.3", "--complete"}, &bytes.Buffer{},
+		func(_ context.Context, _, _ string, complete bool) error {
+			activeChecked = complete
+			return nil
+		},
+		client,
+	)
+
+	// Assert
+	require.NoError(t, err)
+	require.True(t, activeChecked)
+	require.True(t, client.complete)
+}
+
+func TestAppStartAcceptsPreviousUpdaterInvocation(t *testing.T) {
+	// Act
+	err := run(t.Context(), []string{"app-start", "v1.2.3"})
+
+	// Assert
+	require.ErrorContains(t, err, "packaged release")
+}

server/internal/domain/updates/executor_test.go

@@ -133,6 +133,31 @@ func TestUnixExecutorClientTrigger(t *testing.T) {
 	assert.NoError(t, observation.decodeErr)
 }
 
+func TestUnixExecutorClientTriggerComplete(t *testing.T) {
+	// Arrange
+	operationID := "11111111-1111-4111-8111-111111111111"
+	observed := make(chan executorRequestObservation, 1)
+	client := startExecutorTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var request updaterapi.TriggerRequest
+		decodeErr := json.NewDecoder(r.Body).Decode(&request)
+		observed <- executorRequestObservation{trigger: request, decodeErr: decodeErr}
+		w.WriteHeader(http.StatusAccepted)
+		_ = json.NewEncoder(w).Encode(updaterapi.TriggerResponse{Operation: updaterapi.Operation{
+			ID: operationID, TargetVersion: "v1.2.3", Complete: true, Phase: updaterapi.PhaseQueued,
+		}})
+	}))
+
+	// Act
+	operation, err := client.TriggerComplete(t.Context(), operationID, "v1.2.3")
+
+	// Assert
+	require.NoError(t, err)
+	require.Equal(t, operationID, operation.ID)
+	observation := <-observed
+	require.NoError(t, observation.decodeErr)
+	require.True(t, observation.trigger.Complete)
+}
+
 func TestUnixExecutorClientHTTPFailures(t *testing.T) {
 	t.Parallel()
 
@@ -209,6 +234,24 @@ func TestUnixExecutorClientRejectsMismatchedTriggerIdentity(t *testing.T) {
 	assert.ErrorAs(t, err, &protocolErr)
 }
 
+func TestUnixExecutorClientRejectsMismatchedCompletionMode(t *testing.T) {
+	// Arrange
+	client := startExecutorTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusAccepted)
+		_ = json.NewEncoder(w).Encode(updaterapi.TriggerResponse{Operation: updaterapi.Operation{
+			ID: "11111111-1111-4111-8111-111111111111", TargetVersion: "v1.2.3",
+		}})
+	}))
+
+	// Act
+	_, err := client.TriggerComplete(t.Context(), "11111111-1111-4111-8111-111111111111", "v1.2.3")
+
+	// Assert
+	require.Error(t, err)
+	var protocolErr *updaterapi.ProtocolError
+	assert.ErrorAs(t, err, &protocolErr)
+}
+
 func TestUnixExecutorClientUnavailableSocket(t *testing.T) {
 	t.Parallel()