From bab3c7f706545a09d182dd440e91116623729b39 Mon Sep 17 00:00:00 2001 From: Kevin <235441252+KT-Doan@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:37:48 +0900 Subject: [PATCH 1/4] runner: do not force pending_delete from the agent terminated path When a runner reports terminated, SetInstanceToPendingDelete force-updated the instance to pending_delete, bypassing the status transition table. If the provider worker had already moved the instance to deleting and was mid-way through the provider delete call, the forced write moved the record backwards. The provider worker then failed its final transition (pending_delete -> deleted is not a valid transition), and the record was stranded in pending_delete forever while the compute resource was already gone. Such phantom records count toward the scale set's runner count and can permanently wedge autoscaling into scale-down. Use a validated update instead, and treat a rejected transition from a deletion-lane status as success: the intent of the call, the runner goes away, is already being fulfilled. Signed-off-by: Kevin <235441252+KT-Doan@users.noreply.github.com> --- runner/agent.go | 9 ++- runner/agent_test.go | 144 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 152 insertions(+), 1 deletion(-) create mode 100644 runner/agent_test.go diff --git a/runner/agent.go b/runner/agent.go index e8ccdf056..8751d753f 100644 --- a/runner/agent.go +++ b/runner/agent.go @@ -2,6 +2,7 @@ package runner import ( "context" + "errors" "fmt" "log/slog" "time" @@ -9,6 +10,7 @@ import ( runnerErrors "github.com/cloudbase/garm-provider-common/errors" commonParams "github.com/cloudbase/garm-provider-common/params" "github.com/cloudbase/garm/auth" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/params" ) @@ -60,7 +62,12 @@ func (r *Runner) SetInstanceToPendingDelete(ctx context.Context) error { Status: commonParams.InstancePendingDelete, } - if _, err := r.store.ForceUpdateInstance(r.ctx, instance.ID, updateParams); err != nil { + if _, err := r.store.UpdateInstance(r.ctx, instance.ID, updateParams); err != nil { + var te *runnerErrors.InstanceTransitionError + if errors.As(err, &te) && garmErrors.InstanceIsBeingDeleted(te.From) { + // Already on the deletion lane; treat the refusal as success. + return nil + } return fmt.Errorf("failed to set instance to pending_delete: %w", err) } return nil diff --git a/runner/agent_test.go b/runner/agent_test.go new file mode 100644 index 000000000..69f215b60 --- /dev/null +++ b/runner/agent_test.go @@ -0,0 +1,144 @@ +// Copyright 2025 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package runner + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/suite" + + commonParams "github.com/cloudbase/garm-provider-common/params" + "github.com/cloudbase/garm/auth" + "github.com/cloudbase/garm/database" + dbCommon "github.com/cloudbase/garm/database/common" + garmTesting "github.com/cloudbase/garm/internal/testing" + "github.com/cloudbase/garm/params" +) + +type AgentTestSuite struct { + suite.Suite + + adminCtx context.Context + store dbCommon.Store + runner *Runner + instance params.Instance +} + +func (s *AgentTestSuite) SetupTest() { + dbCfg := garmTesting.GetTestSqliteDBConfig(s.T()) + db, err := database.NewDatabase(context.Background(), dbCfg) + if err != nil { + s.FailNow(fmt.Sprintf("failed to create db connection: %s", err)) + } + s.store = db + + s.adminCtx = garmTesting.ImpersonateAdminContext(context.Background(), db, s.T()) + + githubEndpoint := garmTesting.CreateDefaultGithubEndpoint(s.adminCtx, db, s.T()) + testCreds := garmTesting.CreateTestGithubCredentials(s.adminCtx, "test-creds", db, s.T(), githubEndpoint) + + org, err := db.CreateOrganization(s.adminCtx, "test-org", testCreds, "test-webhook-secret", params.PoolBalancerTypeRoundRobin, false) + if err != nil { + s.FailNow(fmt.Sprintf("failed to create test org: %s", err)) + } + + entity, err := org.GetEntity() + if err != nil { + s.FailNow(fmt.Sprintf("failed to get entity: %s", err)) + } + + pool, err := db.CreateEntityPool(s.adminCtx, entity, params.CreatePoolParams{ + ProviderName: "test-provider", + MaxRunners: 2, + MinIdleRunners: 0, + Image: "ubuntu:22.04", + Flavor: "medium", + OSType: commonParams.Linux, + OSArch: commonParams.Amd64, + Tags: []string{"linux", "amd64"}, + RunnerBootstrapTimeout: 10, + }) + if err != nil { + s.FailNow(fmt.Sprintf("failed to create test pool: %s", err)) + } + + instance, err := db.CreateInstance(s.adminCtx, pool.ID, params.CreateInstanceParams{ + Name: "test-agent-instance", + OSType: commonParams.Linux, + OSArch: commonParams.Amd64, + }) + if err != nil { + s.FailNow(fmt.Sprintf("failed to create test instance: %s", err)) + } + s.instance = instance + + s.runner = &Runner{ + ctx: s.adminCtx, + store: db, + } +} + +func (s *AgentTestSuite) seedStatus(status commonParams.InstanceStatus) { + _, err := s.store.ForceUpdateInstance(s.adminCtx, s.instance.Name, params.UpdateInstanceParams{ + Status: status, + }) + s.Require().NoError(err) +} + +func (s *AgentTestSuite) instanceCtx() context.Context { + return auth.SetInstanceParams(context.Background(), s.instance) +} + +// A terminated report on a running instance marks it pending_delete. +func (s *AgentTestSuite) TestSetInstanceToPendingDeleteFromRunning() { + s.seedStatus(commonParams.InstanceRunning) + + err := s.runner.SetInstanceToPendingDelete(s.instanceCtx()) + s.Require().NoError(err) + + updated, err := s.store.GetInstance(s.adminCtx, s.instance.Name) + s.Require().NoError(err) + s.Require().Equal(commonParams.InstancePendingDelete, updated.Status) +} + +// A terminated report must not move a deleting instance backwards. +func (s *AgentTestSuite) TestSetInstanceToPendingDeleteDoesNotRegressDeleting() { + s.seedStatus(commonParams.InstanceDeleting) + + err := s.runner.SetInstanceToPendingDelete(s.instanceCtx()) + s.Require().NoError(err) + + updated, err := s.store.GetInstance(s.adminCtx, s.instance.Name) + s.Require().NoError(err) + s.Require().Equal(commonParams.InstanceDeleting, updated.Status) +} + +// A terminated report on an already deleted record is a no-op. +func (s *AgentTestSuite) TestSetInstanceToPendingDeleteToleratesDeleted() { + s.seedStatus(commonParams.InstanceDeleted) + + err := s.runner.SetInstanceToPendingDelete(s.instanceCtx()) + s.Require().NoError(err) + + updated, err := s.store.GetInstance(s.adminCtx, s.instance.Name) + s.Require().NoError(err) + s.Require().Equal(commonParams.InstanceDeleted, updated.Status) +} + +func TestAgentTestSuite(t *testing.T) { + suite.Run(t, new(AgentTestSuite)) +} From d47f29060b3a6abc4209d86c5802fc9445e0f3a3 Mon Sep 17 00:00:00 2001 From: Kevin <235441252+KT-Doan@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:37:48 +0900 Subject: [PATCH 2/4] workers/provider: resume interrupted deletes and converge on deleted The instance manager's consolidateState had no case for InstanceDeleting. A manager whose cached state landed on deleting (for example when the status update from its own transition was delivered only after a long provider delete call finished, because handleUpdate serializes on the same mutex the delete path holds) ticked every 5 seconds doing nothing, forever. Combined with a racing forced pending_delete write from another worker, this stranded database records in pending_delete with the compute resource already gone: phantom runners that count toward the scale set runner count and wedge autoscaling (#854). Three changes: - Treat InstanceDeleting as a resumable delete. Deletes are idempotent; providers report a missing instance as success. - Requeue on provider errors for every non-forced path, so a resumed delete that fails does not fall through and mark a live resource deleted. - If the final transition to deleted is refused because the row slid back onto the deletion lane (a racing pending_delete write), force the write: the provider resource is confirmed gone at that point. Signed-off-by: Kevin <235441252+KT-Doan@users.noreply.github.com> --- workers/provider/instance_manager.go | 22 ++- workers/provider/instance_manager_test.go | 163 ++++++++++++++++++++++ 2 files changed, 181 insertions(+), 4 deletions(-) create mode 100644 workers/provider/instance_manager_test.go diff --git a/workers/provider/instance_manager.go b/workers/provider/instance_manager.go index f276bd404..f43293d4b 100644 --- a/workers/provider/instance_manager.go +++ b/workers/provider/instance_manager.go @@ -26,6 +26,7 @@ import ( commonParams "github.com/cloudbase/garm-provider-common/params" "github.com/cloudbase/garm/cache" dbCommon "github.com/cloudbase/garm/database/common" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" garmUtil "github.com/cloudbase/garm/util" @@ -339,9 +340,10 @@ func (i *instanceManager) consolidateState() error { } case commonParams.InstanceRunning: // Nothing to do. The provider finished creating the instance. - case commonParams.InstancePendingDelete, commonParams.InstancePendingForceDelete: + case commonParams.InstancePendingDelete, commonParams.InstancePendingForceDelete, commonParams.InstanceDeleting: // Remove or force remove the runner. When force remove is specified, we ignore // IaaS errors. + // InstanceDeleting resumes an unfinished delete; deletes are idempotent. if i.instance.Status == commonParams.InstancePendingDelete { // invoke backoff sleep. We only do this for non forced removals, // as force delete will always return, regardless of whether or not @@ -362,8 +364,8 @@ func (i *instanceManager) consolidateState() error { } if err := i.handleDeleteInstanceInProvider(i.instance); err != nil { - slog.ErrorContext(i.ctx, "deleting instance in provider", "error", err, "forced", i.instance.Status == commonParams.InstancePendingForceDelete) - if prevStatus == commonParams.InstancePendingDelete { + slog.ErrorContext(i.ctx, "deleting instance in provider", "error", err, "forced", prevStatus == commonParams.InstancePendingForceDelete) + if prevStatus != commonParams.InstancePendingForceDelete { i.incrementBackOff() if err := i.helper.SetInstanceStatus(i.instance.Name, commonParams.InstancePendingDelete, []byte(err.Error()), true); err != nil { return fmt.Errorf("setting instance status to error: %w", err) @@ -373,7 +375,19 @@ func (i *instanceManager) consolidateState() error { } } if err := i.helper.SetInstanceStatus(i.instance.Name, commonParams.InstanceDeleted, nil, false); err != nil { - if !errors.Is(err, runnerErrors.ErrNotFound) { + var transitionErr *runnerErrors.InstanceTransitionError + switch { + case errors.Is(err, runnerErrors.ErrNotFound): + // The record is already gone; nothing left to update. + case errors.As(err, &transitionErr) && garmErrors.InstanceIsBeingDeleted(transitionErr.From): + // The row moved back onto the deletion lane; the provider + // resource is gone, so force the final transition. + if err := i.helper.SetInstanceStatus(i.instance.Name, commonParams.InstanceDeleted, nil, true); err != nil { + if !errors.Is(err, runnerErrors.ErrNotFound) { + return fmt.Errorf("setting instance status to deleted: %w", err) + } + } + default: return fmt.Errorf("setting instance status to deleted: %w", err) } } diff --git a/workers/provider/instance_manager_test.go b/workers/provider/instance_manager_test.go new file mode 100644 index 000000000..71f834412 --- /dev/null +++ b/workers/provider/instance_manager_test.go @@ -0,0 +1,163 @@ +// Copyright 2025 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. +package provider + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + runnerErrors "github.com/cloudbase/garm-provider-common/errors" + commonParams "github.com/cloudbase/garm-provider-common/params" + "github.com/cloudbase/garm/auth" + "github.com/cloudbase/garm/params" + "github.com/cloudbase/garm/runner/common" + runnerCommonMocks "github.com/cloudbase/garm/runner/common/mocks" +) + +type statusWrite struct { + status commonParams.InstanceStatus + force bool +} + +// fakeProviderHelper records status writes and lets a test script their +// outcome. Everything else returns zero values; consolidateState's deletion +// path only needs SetInstanceStatus and GetControllerInfo. +type fakeProviderHelper struct { + writes []statusWrite + onSetStatus func(w statusWrite) error +} + +func (f *fakeProviderHelper) SetInstanceStatus(_ string, status commonParams.InstanceStatus, _ []byte, force bool) error { + w := statusWrite{status: status, force: force} + f.writes = append(f.writes, w) + if f.onSetStatus != nil { + return f.onSetStatus(w) + } + return nil +} + +func (f *fakeProviderHelper) InstanceTokenGetter() auth.InstanceTokenGetter { return nil } + +func (f *fakeProviderHelper) updateArgsFromProviderInstance(_ string, _ commonParams.ProviderInstance) (params.Instance, error) { + return params.Instance{}, nil +} + +func (f *fakeProviderHelper) GetControllerInfo() (params.ControllerInfo, error) { + return params.ControllerInfo{}, nil +} + +func (f *fakeProviderHelper) GetGithubEntity(e params.ForgeEntity) (params.ForgeEntity, error) { + return e, nil +} + +func newTestManager(status commonParams.InstanceStatus, provider common.Provider, helper providerHelper) *instanceManager { + m := &instanceManager{ + ctx: context.Background(), + instance: params.Instance{ + Name: "test-instance", + ProviderID: "test-instance", + Status: status, + }, + provider: provider, + helper: helper, + } + m.running.Store(true) + return m +} + +// A manager whose cached state is deleting must resume the delete. +func TestConsolidateStateResumesDeleting(t *testing.T) { + providerMock := runnerCommonMocks.NewProvider(t) + providerMock.On("DeleteInstance", mock.Anything, "test-instance", mock.Anything).Return(nil) + helper := &fakeProviderHelper{} + m := newTestManager(commonParams.InstanceDeleting, providerMock, helper) + + err := m.consolidateState() + require.ErrorIs(t, err, ErrInstanceDeleted) + require.Equal(t, []statusWrite{ + {status: commonParams.InstanceDeleting, force: true}, + {status: commonParams.InstanceDeleted, force: false}, + }, helper.writes) +} + +// A resumed delete that fails in the provider requeues. +func TestConsolidateStateResumedDeletingRequeuesOnProviderError(t *testing.T) { + providerMock := runnerCommonMocks.NewProvider(t) + providerMock.On("DeleteInstance", mock.Anything, "test-instance", mock.Anything).Return(fmt.Errorf("provider exploded")) + helper := &fakeProviderHelper{} + m := newTestManager(commonParams.InstanceDeleting, providerMock, helper) + + err := m.consolidateState() + require.Error(t, err) + require.NotErrorIs(t, err, ErrInstanceDeleted) + require.Equal(t, []statusWrite{ + {status: commonParams.InstanceDeleting, force: true}, + {status: commonParams.InstancePendingDelete, force: true}, + }, helper.writes) + require.Greater(t, m.deleteBackoff.Nanoseconds(), int64(0)) +} + +// Force delete keeps its semantics: provider errors are ignored and the +// instance is marked deleted. +func TestConsolidateStateForceDeleteIgnoresProviderError(t *testing.T) { + providerMock := runnerCommonMocks.NewProvider(t) + providerMock.On("DeleteInstance", mock.Anything, "test-instance", mock.Anything).Return(fmt.Errorf("provider exploded")) + helper := &fakeProviderHelper{} + m := newTestManager(commonParams.InstancePendingForceDelete, providerMock, helper) + + err := m.consolidateState() + require.ErrorIs(t, err, ErrInstanceDeleted) + require.Equal(t, []statusWrite{ + {status: commonParams.InstanceDeleting, force: true}, + {status: commonParams.InstanceDeleted, force: false}, + }, helper.writes) +} + +// A deleted write refused because the row regressed onto the deletion +// lane is retried with force. +func TestConsolidateStateForcesDeletedWhenRowRegressed(t *testing.T) { + providerMock := runnerCommonMocks.NewProvider(t) + providerMock.On("DeleteInstance", mock.Anything, "test-instance", mock.Anything).Return(nil) + helper := &fakeProviderHelper{} + helper.onSetStatus = func(w statusWrite) error { + if w.status == commonParams.InstanceDeleted && !w.force { + return fmt.Errorf("updating instance: %w", runnerErrors.NewInstanceTransitionError(commonParams.InstancePendingDelete, commonParams.InstanceDeleted)) + } + return nil + } + m := newTestManager(commonParams.InstancePendingDelete, providerMock, helper) + + err := m.consolidateState() + require.ErrorIs(t, err, ErrInstanceDeleted) + require.Equal(t, []statusWrite{ + {status: commonParams.InstanceDeleting, force: true}, + {status: commonParams.InstanceDeleted, force: false}, + {status: commonParams.InstanceDeleted, force: true}, + }, helper.writes) +} + +// Running remains a no-op. +func TestConsolidateStateRunningIsNoop(t *testing.T) { + providerMock := runnerCommonMocks.NewProvider(t) + helper := &fakeProviderHelper{} + m := newTestManager(commonParams.InstanceRunning, providerMock, helper) + + err := m.consolidateState() + require.NoError(t, err) + require.Empty(t, helper.writes) +} From a7684d739e3f34b49c690cb88ac6e5c8d7ffa4f2 Mon Sep 17 00:00:00 2001 From: Kevin <235441252+KT-Doan@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:37:49 +0900 Subject: [PATCH 3/4] workers/scaleset: exclude deletion-lane records from the runner count runnerCount() was len(w.runners), which counts records in pending_delete, pending_force_delete, deleting and deleted. Those records cannot pick up jobs. Counting them suppresses scale-up, and a record stuck on the deletion lane (see the phantom pending_delete wedge, #854) pinned the autoscaler in permanent scale-down: with min_idle_runners 0 and desired usually 1 on a quiet set, two stuck records were enough to starve any scale set regardless of its max_runners. handleScaleDown's deletion-lane case no longer consumes the removal delta for the same reason: those records are no longer part of the count the delta was derived from. Signed-off-by: Kevin <235441252+KT-Doan@users.noreply.github.com> --- workers/scaleset/runner_count_test.go | 59 +++++++++++++++++++++++++++ workers/scaleset/scaleset.go | 12 +++++- 2 files changed, 69 insertions(+), 2 deletions(-) create mode 100644 workers/scaleset/runner_count_test.go diff --git a/workers/scaleset/runner_count_test.go b/workers/scaleset/runner_count_test.go new file mode 100644 index 000000000..18afce181 --- /dev/null +++ b/workers/scaleset/runner_count_test.go @@ -0,0 +1,59 @@ +// Copyright 2025 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. +package scaleset + +import ( + "testing" + + "github.com/stretchr/testify/require" + + commonParams "github.com/cloudbase/garm-provider-common/params" + "github.com/cloudbase/garm/params" +) + +// Deletion-lane records do not count toward the runner count, so a scale +// set holding only such records still scales up. +func TestRunnerCountExcludesDeletionLane(t *testing.T) { + w := &Worker{ + scaleSet: params.ScaleSet{ + MaxRunners: 48, + MinIdleRunners: 0, + DesiredRunnerCount: 1, + }, + runners: map[string]params.Instance{ + "a": {Name: "phantom-1", Status: commonParams.InstancePendingDelete}, + "b": {Name: "phantom-2", Status: commonParams.InstancePendingDelete}, + }, + } + + require.Equal(t, 0, w.runnerCount()) + require.Less(t, w.runnerCount(), w.targetRunners()) +} + +func TestRunnerCountMixedStatuses(t *testing.T) { + w := &Worker{ + runners: map[string]params.Instance{ + "a": {Status: commonParams.InstanceRunning}, + "b": {Status: commonParams.InstancePendingCreate}, + "c": {Status: commonParams.InstanceCreating}, + "d": {Status: commonParams.InstanceError}, + "e": {Status: commonParams.InstancePendingDelete}, + "f": {Status: commonParams.InstancePendingForceDelete}, + "g": {Status: commonParams.InstanceDeleting}, + "h": {Status: commonParams.InstanceDeleted}, + }, + } + + require.Equal(t, 4, w.runnerCount()) +} diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 418756c0d..a642a949b 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -1005,7 +1005,7 @@ func (w *Worker) handleScaleDown() { removed++ case commonParams.InstancePendingDelete, commonParams.InstancePendingForceDelete, commonParams.InstanceDeleting, commonParams.InstanceDeleted: - removed++ + // Not counted by runnerCount(), so not part of the delta. continue default: slog.WarnContext(w.ctx, "runner is not in a valid state; skipping", "runner_name", runner.Name, "runner_status", runner.Status) @@ -1024,8 +1024,16 @@ func (w *Worker) targetRunners() int { return int(targetRunners) } +// runnerCount counts runners that are not on the deletion lane. func (w *Worker) runnerCount() int { - return len(w.runners) + count := 0 + for _, runner := range w.runners { + if garmErrors.InstanceIsBeingDeleted(runner.Status) { + continue + } + count++ + } + return count } func (w *Worker) handleAutoScale() { From bd0c265f7d50ab1685cf5df51d2f4399d9e5bc06 Mon Sep 17 00:00:00 2001 From: Kevin <235441252+KT-Doan@users.noreply.github.com> Date: Thu, 20 Aug 2026 15:27:02 +0900 Subject: [PATCH 4/4] workers/provider: add end-to-end regression test for the pending_delete wedge Replays the full incident interleave from #854 through a real store, the real watcher, and a running provider worker: a running instance moves to pending_delete, the manager starts a slow provider delete holding its mutex, the deleting event parks inside handleUpdate, a forced pending_delete write regresses the row mid-delete, and the follow-up event is dropped by Update()'s 10s send timeout. Without the fixes in this series the manager's cached state lands on deleting, a state consolidateState did not handle, and the record stays in pending_delete forever (this test times out on main). With them the manager re-drives the delete and the record converges to deleted within a couple of ticks. Signed-off-by: Kevin <235441252+KT-Doan@users.noreply.github.com> --- workers/provider/wedge_regression_test.go | 163 ++++++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 workers/provider/wedge_regression_test.go diff --git a/workers/provider/wedge_regression_test.go b/workers/provider/wedge_regression_test.go new file mode 100644 index 000000000..98d9693bf --- /dev/null +++ b/workers/provider/wedge_regression_test.go @@ -0,0 +1,163 @@ +// Copyright 2025 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. +package provider + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + commonParams "github.com/cloudbase/garm-provider-common/params" + "github.com/cloudbase/garm/database" + "github.com/cloudbase/garm/database/watcher" + garmTesting "github.com/cloudbase/garm/internal/testing" + "github.com/cloudbase/garm/params" + "github.com/cloudbase/garm/runner/common" + runnerCommonMocks "github.com/cloudbase/garm/runner/common/mocks" +) + +type wedgeTokenGetter struct{} + +func (wedgeTokenGetter) NewInstanceJWTToken(_ params.Instance, _ params.ForgeEntity, _ uint) (string, error) { + return "test-token", nil +} + +func (wedgeTokenGetter) NewAgentJWTToken(_ params.Instance, _ params.ForgeEntity) (string, error) { + return "test-token", nil +} + +// A slow provider delete is raced by a forced pending_delete write, and +// the manager's own status update arrives only after the delete returns. +// The record must still converge to deleted instead of staying in +// pending_delete. +func TestPendingDeleteClobberWedgeConverges(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + watcher.InitWatcher(ctx) + t.Cleanup(func() { watcher.CloseWatcher() }) + + dbCfg := garmTesting.GetTestSqliteDBConfig(t) + db, err := database.NewDatabase(ctx, dbCfg) + require.NoError(t, err) + + adminCtx := garmTesting.ImpersonateAdminContext(ctx, db, t) + + _, err = db.InitController() + require.NoError(t, err) + + githubEndpoint := garmTesting.CreateDefaultGithubEndpoint(adminCtx, db, t) + testCreds := garmTesting.CreateTestGithubCredentials(adminCtx, "test-creds", db, t, githubEndpoint) + + org, err := db.CreateOrganization(adminCtx, "test-org", testCreds, "test-webhook-secret", params.PoolBalancerTypeRoundRobin, false) + require.NoError(t, err) + entity, err := org.GetEntity() + require.NoError(t, err) + + scaleSet, err := db.CreateEntityScaleSet(adminCtx, entity, params.CreateScaleSetParams{ + Name: "test-scale-set", + ScaleSetID: 1, + ProviderName: "test-provider", + MaxRunners: 5, + MinIdleRunners: 0, + Image: "ubuntu:22.04", + Flavor: "medium", + OSType: commonParams.Linux, + OSArch: commonParams.Amd64, + Enabled: true, + RunnerBootstrapTimeout: 10, + }) + require.NoError(t, err) + + instance, err := db.CreateScaleSetInstance(adminCtx, scaleSet.ID, params.CreateInstanceParams{ + Name: "test-wedge-instance", + Status: commonParams.InstancePendingCreate, + OSType: commonParams.Linux, + OSArch: commonParams.Amd64, + }) + require.NoError(t, err) + _, err = db.ForceUpdateInstance(adminCtx, instance.Name, params.UpdateInstanceParams{ + Status: commonParams.InstanceRunning, + }) + require.NoError(t, err) + + deleteEntered := make(chan struct{}) + releaseDelete := make(chan struct{}) + providerMock := runnerCommonMocks.NewProvider(t) + // The first delete call signals the test and blocks until released; + // later calls return immediately. + providerMock.On("DeleteInstance", mock.Anything, instance.Name, mock.Anything).Run(func(_ mock.Arguments) { + close(deleteEntered) + <-releaseDelete + }).Return(nil).Once() + providerMock.On("DeleteInstance", mock.Anything, instance.Name, mock.Anything).Return(nil).Maybe() + + worker, err := NewWorker(ctx, db, map[string]common.Provider{"test-provider": providerMock}, wedgeTokenGetter{}) + require.NoError(t, err) + require.NoError(t, worker.Start()) + t.Cleanup(func() { + if err := worker.Stop(); err != nil { + t.Logf("stopping worker: %v", err) + } + }) + + // running -> pending_delete, as on job completion. + _, err = db.UpdateInstance(adminCtx, instance.Name, params.UpdateInstanceParams{ + Status: commonParams.InstancePendingDelete, + }) + require.NoError(t, err) + + // Wait for the manager to start the provider delete. + select { + case <-deleteEntered: + case <-time.After(30 * time.Second): + t.Fatal("provider delete was never called") + } + + // Let the deleting event reach the manager before the write below; the + // watcher does not guarantee delivery order. + time.Sleep(2 * time.Second) + + // Force pending_delete over deleting while the delete is in flight. + _, err = db.ForceUpdateInstance(adminCtx, instance.Name, params.UpdateInstanceParams{ + Status: commonParams.InstancePendingDelete, + }) + require.NoError(t, err) + regressed, err := db.GetInstance(adminCtx, instance.Name) + require.NoError(t, err) + require.Equal(t, commonParams.InstancePendingDelete, regressed.Status) + + // Hold the delete past the 10s update-send timeout so the second event + // is dropped, then let it finish. + time.Sleep(12 * time.Second) + close(releaseDelete) + + // The record must converge to deleted. + require.Eventually(t, func() bool { + inst, err := db.GetInstance(adminCtx, instance.Name) + if err != nil { + // The record was fully removed; also convergence. + return true + } + return inst.Status == commonParams.InstanceDeleted + }, 60*time.Second, time.Second, "instance never converged to deleted") + + if inst, err := db.GetInstance(adminCtx, instance.Name); err == nil { + fmt.Printf("converged instance status: %s\n", inst.Status) + } +}