diff --git a/internal/command/deploy/machine_update_outcome.go b/internal/command/deploy/machine_update_outcome.go new file mode 100644 index 0000000000..9c36a2e6a7 --- /dev/null +++ b/internal/command/deploy/machine_update_outcome.go @@ -0,0 +1,90 @@ +package deploy + +import ( + "context" + "time" + + fly "github.com/superfly/fly-go" + "github.com/superfly/flyctl/internal/machine" +) + +func supportsPreservedStoppedUpdate(strategy, launchBasisState string) bool { + if strategy != "canary" && strategy != "rolling" { + return false + } + + return launchBasisState == fly.MachineStateStarted || launchBasisState == "starting" +} + +func isPreservedStoppedUpdate(current *fly.Machine, updatedInstanceID string) bool { + if current == nil || current.Config == nil || current.Config.Schedule != "" || + current.State != fly.MachineStateStopped || current.InstanceID != updatedInstanceID { + return false + } + + for _, event := range current.Events { + if event == nil || event.Status == "" { + continue + } + + return event.Type == "update" && event.Status == fly.MachineStateStopped && event.Source == "flyd" + } + + return false +} + +func (md *machineDeployment) readPreservedStoppedUpdate(ctx context.Context, machineID, updatedInstanceID string) bool { + current, err := md.flapsClient.Get(ctx, md.app.Name, machineID) + + return err == nil && isPreservedStoppedUpdate(current, updatedInstanceID) +} + +func (md *machineDeployment) waitForStartedOrPreservedStoppedUpdate( + ctx context.Context, + lm machine.LeasableMachine, + launchBasisState string, + timeout time.Duration, +) (bool, error) { + if !supportsPreservedStoppedUpdate(md.strategy, launchBasisState) { + return false, lm.WaitForState(ctx, fly.MachineStateStarted, timeout, machine.WithJustCreated()) + } + + waitCtx, cancel := context.WithCancel(ctx) + defer cancel() + + startedResult := make(chan error, 1) + go func() { + startedResult <- lm.WaitForState(waitCtx, fly.MachineStateStarted, timeout, machine.WithJustCreated()) + }() + + stoppedResult := make(chan error, 1) + go func() { + stoppedResult <- lm.WaitForState(waitCtx, fly.MachineStateStopped, timeout, machine.WithJustCreated()) + }() + + updated := lm.Machine() + var startedErr error + for { + select { + case err := <-startedResult: + startedResult = nil + if err == nil { + return false, nil + } + startedErr = err + if stoppedResult == nil { + return false, startedErr + } + case err := <-stoppedResult: + stoppedResult = nil + if err == nil && md.readPreservedStoppedUpdate(waitCtx, updated.ID, updated.InstanceID) { + return true, nil + } + if startedResult == nil { + return false, startedErr + } + case <-waitCtx.Done(): + return false, waitCtx.Err() + } + } +} diff --git a/internal/command/deploy/machine_update_outcome_test.go b/internal/command/deploy/machine_update_outcome_test.go new file mode 100644 index 0000000000..c05ff5b07a --- /dev/null +++ b/internal/command/deploy/machine_update_outcome_test.go @@ -0,0 +1,293 @@ +package deploy + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + fly "github.com/superfly/fly-go" + "github.com/superfly/fly-go/flaps" + "github.com/superfly/flyctl/internal/appconfig" + "github.com/superfly/flyctl/internal/machine" + "github.com/superfly/flyctl/internal/statuslogger" + "github.com/superfly/flyctl/iostreams" +) + +type updateOutcomeRoundTripFunc func(*http.Request) (*http.Response, error) + +func (f updateOutcomeRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +func TestIsPreservedStoppedUpdate(t *testing.T) { + preserved := &fly.MachineEvent{Type: "update", Status: fly.MachineStateStopped, Source: "flyd"} + + tests := []struct { + name string + machine *fly.Machine + instance string + want bool + }{ + { + name: "matching current update event", + machine: &fly.Machine{ + State: fly.MachineStateStopped, + InstanceID: "new-version", + Config: &fly.MachineConfig{}, + Events: []*fly.MachineEvent{preserved}, + }, + instance: "new-version", + want: true, + }, + { + name: "missing machine declines", + instance: "new-version", + }, + { + name: "missing config declines", + machine: &fly.Machine{ + State: fly.MachineStateStopped, + InstanceID: "new-version", + Events: []*fly.MachineEvent{preserved}, + }, + instance: "new-version", + }, + { + name: "scheduled update declines", + machine: &fly.Machine{ + State: fly.MachineStateStopped, + InstanceID: "new-version", + Config: &fly.MachineConfig{Schedule: "daily"}, + Events: []*fly.MachineEvent{preserved}, + }, + instance: "new-version", + }, + { + name: "newer lifecycle event declines", + machine: &fly.Machine{ + State: fly.MachineStateStopped, + InstanceID: "new-version", + Config: &fly.MachineConfig{}, + Events: []*fly.MachineEvent{ + {Type: "exit", Status: fly.MachineStateStopped, Source: "flyd"}, + preserved, + }, + }, + instance: "new-version", + }, + { + name: "different version declines", + machine: &fly.Machine{ + State: fly.MachineStateStopped, + InstanceID: "other-version", + Config: &fly.MachineConfig{}, + Events: []*fly.MachineEvent{preserved}, + }, + instance: "new-version", + }, + { + name: "non-flyd event declines", + machine: &fly.Machine{ + State: fly.MachineStateStopped, + InstanceID: "new-version", + Config: &fly.MachineConfig{}, + Events: []*fly.MachineEvent{ + {Type: "update", Status: fly.MachineStateStopped, Source: "user"}, + }, + }, + instance: "new-version", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isPreservedStoppedUpdate(tc.machine, tc.instance)) + }) + } +} + +func TestSupportsPreservedStoppedUpdate(t *testing.T) { + require.True(t, supportsPreservedStoppedUpdate("canary", fly.MachineStateStarted)) + require.True(t, supportsPreservedStoppedUpdate("rolling", "starting")) + require.False(t, supportsPreservedStoppedUpdate("canary", "failed")) + require.False(t, supportsPreservedStoppedUpdate("immediate", fly.MachineStateStarted)) +} + +func TestWaitForMachineAcceptsPreservedStoppedUpdate(t *testing.T) { + t.Setenv("FLY_FLAPS_BASE_URL", "http://flaps.test") + + ios, _, _, _ := iostreams.Test() + const instanceID = "01G6R2TQGS41MBQTCA55X8ZCZW" + startedWaitObserved := make(chan struct{}) + var sawMachineGet atomic.Bool + client, err := flaps.NewWithOptions(context.Background(), flaps.NewClientOpts{ + Transport: updateOutcomeRoundTripFunc(func(req *http.Request) (*http.Response, error) { + state := req.URL.Query().Get("state") + if state == fly.MachineStateStarted { + close(startedWaitObserved) + <-req.Context().Done() + + return nil, req.Context().Err() + } + if state == fly.MachineStateStopped { + select { + case <-startedWaitObserved: + case <-req.Context().Done(): + return nil, req.Context().Err() + } + + return machineResponse(req, ""), nil + } + + sawMachineGet.Store(true) + + return machineResponse(req, `{"id":"machine-id","instance_id":"`+instanceID+`","state":"stopped","events":[{"type":"update","status":"stopped","source":"flyd"}],"config":{}}`), nil + }), + }) + require.NoError(t, err) + entry := &machineUpdateEntry{ + leasableMachine: machine.NewLeasableMachine(client, ios, "app", &fly.Machine{ + ID: "machine-id", + InstanceID: instanceID, + }, false), + launchInput: &fly.LaunchMachineInput{}, + } + md := &machineDeployment{ + app: &flaps.App{Name: "app"}, + io: ios, + flapsClient: client, + strategy: "canary", + waitTimeout: time.Second, + skipSmokeChecks: true, + } + ctx := iostreams.NewContext(context.Background(), ios) + line := statuslogger.Create(ctx, 1, false).Line(0) + + require.NoError(t, md.waitForMachine(ctx, entry, fly.MachineStateStarted, line)) + require.True(t, sawMachineGet.Load()) +} + +func TestWaitForStartedOrPreservedStoppedUpdateDeclinesNewerLifecycleEvent(t *testing.T) { + t.Setenv("FLY_FLAPS_BASE_URL", "http://flaps.test") + + ios, _, _, _ := iostreams.Test() + client, err := flaps.NewWithOptions(context.Background(), flaps.NewClientOpts{ + Transport: updateOutcomeRoundTripFunc(func(req *http.Request) (*http.Response, error) { + switch req.URL.Query().Get("state") { + case fly.MachineStateStarted: + <-req.Context().Done() + + return nil, req.Context().Err() + case fly.MachineStateStopped: + return machineResponse(req, ""), nil + default: + return machineResponse(req, `{"id":"machine-id","instance_id":"new-version","state":"stopped","events":[{"type":"exit","status":"stopped","source":"flyd"},{"type":"update","status":"stopped","source":"flyd"}],"config":{}}`), nil + } + }), + }) + require.NoError(t, err) + md := &machineDeployment{ + app: &flaps.App{Name: "app"}, + flapsClient: client, + strategy: "canary", + } + lm := machine.NewLeasableMachine(client, ios, "app", &fly.Machine{ + ID: "machine-id", + InstanceID: "new-version", + }, false) + + preservedStopped, err := md.waitForStartedOrPreservedStoppedUpdate( + context.Background(), + lm, + fly.MachineStateStarted, + 20*time.Millisecond, + ) + + require.False(t, preservedStopped) + require.Error(t, err) +} + +func TestUpdateMachineWChecksAcceptsPreservedStoppedUpdate(t *testing.T) { + for _, strategy := range []string{"canary", "rolling"} { + t.Run(strategy, func(t *testing.T) { + testUpdateMachineWChecksAcceptsPreservedStoppedUpdate(t, strategy) + }) + } +} + +func testUpdateMachineWChecksAcceptsPreservedStoppedUpdate(t *testing.T, strategy string) { + t.Setenv("FLY_FLAPS_BASE_URL", "http://flaps.test") + + ios, _, _, _ := iostreams.Test() + ctx := iostreams.NewContext(context.Background(), ios) + const instanceID = "01G6R2TQGS41MBQTCA55X8ZCZW" + oldMachine := &fly.Machine{ + ID: "machine-id", + State: fly.MachineStateStarted, + LeaseNonce: "lease-nonce", + HostStatus: fly.HostStatusOk, + Config: &fly.MachineConfig{Image: "image-v1"}, + } + newMachine := &fly.Machine{ + ID: oldMachine.ID, + State: fly.MachineStateStarted, + HostStatus: fly.HostStatusOk, + Config: &fly.MachineConfig{Image: "image-v2"}, + } + + var sawSkipLaunch, sawMachineGet atomic.Bool + client, err := flaps.NewWithOptions(context.Background(), flaps.NewClientOpts{ + Transport: updateOutcomeRoundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.Method == http.MethodPost { + var input fly.LaunchMachineInput + require.NoError(t, json.NewDecoder(req.Body).Decode(&input)) + sawSkipLaunch.Store(input.SkipLaunch) + + return machineResponse(req, `{"id":"machine-id","instance_id":"`+instanceID+`","state":"created","config":{"image":"image-v2"}}`), nil + } + + state := req.URL.Query().Get("state") + if state != "" { + if state == fly.MachineStateStopped { + return machineResponse(req, ""), nil + } + <-req.Context().Done() + + return nil, req.Context().Err() + } + + sawMachineGet.Store(true) + + return machineResponse(req, `{"id":"machine-id","instance_id":"`+instanceID+`","state":"stopped","events":[{"type":"update","status":"stopped","source":"flyd"}],"config":{"image":"image-v2"}}`), nil + }), + }) + require.NoError(t, err) + md := &machineDeployment{ + app: &flaps.App{Name: "app"}, + appConfig: &appconfig.Config{AppName: "app"}, + flapsClient: client, + io: ios, + strategy: strategy, + waitTimeout: 2 * time.Second, + } + line := statuslogger.Create(ctx, 1, false).Line(0) + + require.NoError(t, md.updateMachineWChecks(ctx, oldMachine, newMachine, false, line, ios, &healthcheckResult{})) + require.False(t, sawSkipLaunch.Load(), "observed outcome must not rewrite retry-stable launch intent") + require.True(t, sawMachineGet.Load()) +} + +func machineResponse(req *http.Request, body string) *http.Response { + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + Request: req, + } +} diff --git a/internal/command/deploy/machines_deploymachinesapp.go b/internal/command/deploy/machines_deploymachinesapp.go index e4368d12b3..55297ba081 100644 --- a/internal/command/deploy/machines_deploymachinesapp.go +++ b/internal/command/deploy/machines_deploymachinesapp.go @@ -207,7 +207,7 @@ func (md *machineDeployment) updateMachine(ctx context.Context, e *machineUpdate return nil } -func (md *machineDeployment) waitForMachine(ctx context.Context, e *machineUpdateEntry, sl statuslogger.StatusLine) error { +func (md *machineDeployment) waitForMachine(ctx context.Context, e *machineUpdateEntry, launchBasisState string, sl statuslogger.StatusLine) error { lm := e.leasableMachine // Don't wait for SkipLaunch machines, they are updated but not started if e.launchInput.SkipLaunch { @@ -215,11 +215,17 @@ func (md *machineDeployment) waitForMachine(ctx context.Context, e *machineUpdat } if !md.skipHealthChecks { - if err := lm.WaitForState(ctx, fly.MachineStateStarted, md.waitTimeout, machine.WithJustCreated()); err != nil { + preservedStopped, err := md.waitForStartedOrPreservedStoppedUpdate(ctx, lm, launchBasisState, md.waitTimeout) + if err != nil { err = suggestChangeWaitTimeout(err, "wait-timeout") return err } + if preservedStopped { + sl.LogStatus(statuslogger.StatusSuccess, fmt.Sprintf("Machine %s was updated and left stopped", lm.Machine().ID)) + + return nil + } if err := md.runTestMachines(ctx, e.leasableMachine.Machine(), sl); err != nil { return err @@ -1001,13 +1007,14 @@ func (md *machineDeployment) updateEntriesGroup(parentCtx context.Context, group statusRunning() } + launchBasisState := e.leasableMachine.Machine().State if err := md.updateMachine(ctx, e, sl.Line(startIdx+idx)); err != nil { statusFailure(err) tracing.RecordError(span, err, "failed to update machine") return err } - if err := md.waitForMachine(ctx, e, sl.Line(startIdx+idx)); err != nil { + if err := md.waitForMachine(ctx, e, launchBasisState, sl.Line(startIdx+idx)); err != nil { tracing.RecordError(span, err, "failed to wait for machine") statusFailure(err) diff --git a/internal/command/deploy/plan.go b/internal/command/deploy/plan.go index 408a1dc505..b87a80e206 100644 --- a/internal/command/deploy/plan.go +++ b/internal/command/deploy/plan.go @@ -610,12 +610,19 @@ func (md *machineDeployment) updateMachineWChecks(ctx context.Context, oldMachin if !healthcheckResult.machineChecksPassed || !healthcheckResult.smokeChecksPassed { sl.LogStatus(statuslogger.StatusRunning, fmt.Sprintf("Waiting for machine %s to reach a good state", machine.ID)) - _, err := waitForMachineState(ctx, lm, []string{"stopped", "started", "suspended"}, md.waitTimeout, sl) + state, err := waitForMachineState(ctx, lm, []string{"stopped", "started", "suspended"}, md.waitTimeout, sl) if err != nil { span.RecordError(err) return err } + if state == fly.MachineStateStopped && oldMachine != nil && + supportsPreservedStoppedUpdate(md.strategy, oldMachine.State) && + md.readPreservedStoppedUpdate(ctx, machine.ID, machine.InstanceID) { + sl.LogStatus(statuslogger.StatusSuccess, fmt.Sprintf("Machine %s was updated and left stopped", machine.ID)) + + return nil + } } if skipLaunch {