From 412d4a41438ed2615703983bb02485d35d4e77a2 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Thu, 24 Sep 2026 08:41:11 +0000 Subject: [PATCH 1/3] fix: stop leaking a branch worker when nothing needs it any more MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ReconcileWorkers` has been in the WorkerManager since the beginning, does exactly the right thing, and has never been called. Neither has `RegisterTarget`, and `UnregisterTarget` only from tests. The single live creation path is `EnsureWorker` from the GitTarget reconcile, so a worker was created on demand and then lived until the process restarted — holding a goroutine, an event queue and an on-disk clone under /tmp for a GitTarget that may have been deleted hours ago. `cleanupDeletedGitTarget` tore down the event stream and the watch-plane declaration and said nothing about the worker. The sweep now runs from that same cleanup, which is the reconcile the delete watch event produces: the one moment the set of needed workers can have shrunk. It decides from the API rather than from a count of registrations, and that is what makes it right for a branch SHARED by several GitTargets — deleting one leaves the others listed, so the worker they share is left alone. A List that fails stops nothing: treating "I could not read the targets" as "there are no targets" would take down every live worker in the process. `RegisterTarget` and `UnregisterTarget` go. The first had no callers at all. The second destroyed the worker on ANY unregister, siblings or not — which the sweep makes both unnecessary and wrong — and its lock discipline, the part worth keeping, moves into `removeWorkers`: detach under `mu`, stop outside it, hold `lifecycleMu` across, so the queue-depth gauge is never waiting on a worker shutdown and no replacement can start on a clone that is still draining. The tests that drove the deleted API now drive the real one, including the two cases nothing covered before: a sibling keeps the worker alive, and a failed read stops nothing. Co-Authored-By: Claude Opus 5 --- .coverage-baseline | 2 +- .../gittarget_branch_target_metric_test.go | 7 +- internal/controller/gittarget_controller.go | 20 +- .../gittarget_worker_cleanup_test.go | 93 ++++++++++ .../git/branch_worker_queue_depth_test.go | 19 +- internal/git/worker_manager.go | 121 +++++------- internal/git/worker_manager_test.go | 172 ++++++++---------- 7 files changed, 254 insertions(+), 180 deletions(-) create mode 100644 internal/controller/gittarget_worker_cleanup_test.go diff --git a/.coverage-baseline b/.coverage-baseline index 106c8e6c..9274442d 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -78.7 +78.9 diff --git a/internal/controller/gittarget_branch_target_metric_test.go b/internal/controller/gittarget_branch_target_metric_test.go index 279f3741..de0b9b14 100644 --- a/internal/controller/gittarget_branch_target_metric_test.go +++ b/internal/controller/gittarget_branch_target_metric_test.go @@ -3,6 +3,7 @@ package controller import ( + "context" "testing" "github.com/stretchr/testify/assert" @@ -38,7 +39,11 @@ func TestGitTargetCleanup_ForgetsTheBranchTargetMapping(t *testing.T) { require.True(t, ok, "precondition: the mapping is published") r := &GitTargetReconciler{} - r.cleanupDeletedGitTarget(k8stypes.NamespacedName{Namespace: "team-a", Name: "mirror"}, logf.Log) + r.cleanupDeletedGitTarget( + context.Background(), + k8stypes.NamespacedName{Namespace: "team-a", Name: "mirror"}, + logf.Log, + ) _, ok = telemetry.CollectInt64Sum(reader, branchTargetsMetricName, map[string]string{ "gittarget_namespace": "team-a", diff --git a/internal/controller/gittarget_controller.go b/internal/controller/gittarget_controller.go index fd6e0655..2adab12c 100644 --- a/internal/controller/gittarget_controller.go +++ b/internal/controller/gittarget_controller.go @@ -167,7 +167,7 @@ func (r *GitTargetReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( var target configbutleraiv1alpha3.GitTarget if err := r.Get(ctx, req.NamespacedName, &target); err != nil { - return r.handleFetchError(err, log, req.NamespacedName) + return r.handleFetchError(ctx, err, log, req.NamespacedName) } st := beginStatus(r.Client, r.Recorder, &target) @@ -1141,12 +1141,13 @@ func hasAgeKeyEntry(data map[string][]byte) bool { // handleFetchError handles errors from fetching GitTarget. func (r *GitTargetReconciler) handleFetchError( + ctx context.Context, err error, log logr.Logger, namespacedName k8stypes.NamespacedName, ) (ctrl.Result, error) { if client.IgnoreNotFound(err) == nil { - r.cleanupDeletedGitTarget(namespacedName, log) + r.cleanupDeletedGitTarget(ctx, namespacedName, log) log.Info("GitTarget not found, was likely deleted", "namespacedName", namespacedName) return ctrl.Result{}, nil } @@ -1155,6 +1156,7 @@ func (r *GitTargetReconciler) handleFetchError( } func (r *GitTargetReconciler) cleanupDeletedGitTarget( + ctx context.Context, namespacedName k8stypes.NamespacedName, log logr.Logger, ) { @@ -1169,6 +1171,20 @@ func (r *GitTargetReconciler) cleanupDeletedGitTarget( // branch's push failures to an object that no longer exists. telemetry.ForgetBranchTarget(namespacedName.Namespace, namespacedName.Name) + // The branch worker outlives its GitTarget unless something says otherwise, and this reconcile + // — the one the delete watch event produced — is the moment the set of needed workers can have + // shrunk. The sweep decides from the API, so a branch several GitTargets share keeps its worker + // while any of them is still listed. It runs ahead of the EventRouter check below because it is + // the WorkerManager's business, not the data plane's. + if r.WorkerManager != nil { + if err := r.WorkerManager.ReconcileWorkers(ctx); err != nil { + // Not an error the reconcile can act on: the object is already gone, so there is + // nothing to requeue for. The next deletion sweeps again. + log.V(1).Info("Could not sweep branch workers after a GitTarget was deleted", + "error", err.Error()) + } + } + if r.EventRouter == nil { return } diff --git a/internal/controller/gittarget_worker_cleanup_test.go b/internal/controller/gittarget_worker_cleanup_test.go new file mode 100644 index 00000000..ea869acd --- /dev/null +++ b/internal/controller/gittarget_worker_cleanup_test.go @@ -0,0 +1,93 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + k8stypes "k8s.io/apimachinery/pkg/types" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + configbutleraiv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/git" + "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// TestCleanupDeletedGitTarget_StopsAWorkerNothingNeedsAnyMore is the wiring, and the wiring is the +// whole bug: the sweep existed and was never called, so a branch worker — its goroutine, its event +// queue and its on-disk clone — outlived every GitTarget that justified it, until the process +// restarted. The reconcile that the delete watch event produces is where the set of needed workers +// can have shrunk, so that is where the sweep belongs. +func TestCleanupDeletedGitTarget_StopsAWorkerNothingNeedsAnyMore(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") + require.True(t, exists) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + r.cleanupDeletedGitTarget(ctx, + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + + _, exists = workers.GetWorkerForTarget("repo1", "shop", "main") + assert.False(t, exists, "the GitTarget is gone and nothing else lists this branch") +} + +// TestCleanupDeletedGitTarget_LeavesAWorkerAnotherGitTargetStillUses. The sweep reads the API +// rather than counting registrations, and this is why: one worker serves every GitTarget on its +// (provider, branch), so one deletion says nothing about the rest. +func TestCleanupDeletedGitTarget_LeavesAWorkerAnotherGitTargetStillUses(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + + sibling := &configbutleraiv1alpha3.GitTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "invoices", Namespace: "shop"}, + } + sibling.Spec.GitProviderRef.Name = "repo1" + sibling.Spec.Branch = "main" + k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(sibling).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + r.cleanupDeletedGitTarget(ctx, + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + + _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") + assert.True(t, exists, "invoices still writes to this branch") +} + +// TestCleanupDeletedGitTarget_WithoutAWorkerManager covers the CLI and the tests: the cleanup runs +// on a reconciler with no data plane wired, and must not reach for one. +func TestCleanupDeletedGitTarget_WithoutAWorkerManager(t *testing.T) { + r := &GitTargetReconciler{} + assert.NotPanics(t, func() { + r.cleanupDeletedGitTarget(context.Background(), + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + }) +} diff --git a/internal/git/branch_worker_queue_depth_test.go b/internal/git/branch_worker_queue_depth_test.go index 264f27dc..dc6617ea 100644 --- a/internal/git/branch_worker_queue_depth_test.go +++ b/internal/git/branch_worker_queue_depth_test.go @@ -40,7 +40,7 @@ func TestReleaseHandledItem_PublishesRetainedWorkBeforeReleasingInflight(t *test // The git_queue_depth source must not wait on the worker lifecycle lock. // -// queueDepthSamples runs on the metric SDK's collection goroutine. UnregisterTarget used to hold +// queueDepthSamples runs on the metric SDK's collection goroutine. Worker removal used to hold // m.mu across worker.Stop(), which waits for the loop goroutine to finish, so a slow shutdown // blocked collection for its whole duration — the staleness the observable gauge exists to remove, // reintroduced through the manager's own lock. @@ -58,16 +58,17 @@ func TestQueueDepthSamples_NotBlockedByASlowWorkerShutdown(t *testing.T) { release := func() { once.Do(w.wg.Done) } t.Cleanup(release) - var unregistered sync.WaitGroup - unregistered.Add(1) + var removed sync.WaitGroup + removed.Add(1) go func() { - defer unregistered.Done() - _ = manager.UnregisterTarget("", "", "test-provider", "test-ns", "main") + defer removed.Done() + manager.removeWorkers([]BranchKey{{RepoNamespace: "test-ns", RepoName: "test-provider", Branch: "main"}}, + "test") }() sampled := make(chan int, 1) go func() { - // Give the unregister a moment to reach Stop() before sampling. + // Give the removal a moment to reach Stop() before sampling. time.Sleep(50 * time.Millisecond) sampled <- len(manager.queueDepthSamples()) }() @@ -79,7 +80,7 @@ func TestQueueDepthSamples_NotBlockedByASlowWorkerShutdown(t *testing.T) { } release() - unregistered.Wait() + removed.Wait() } // A replacement worker must not start while the one it replaces is still stopping. @@ -120,10 +121,10 @@ func TestEnsureWorker_WaitsForTheWorkerItReplacesToStop(t *testing.T) { stopping := make(chan struct{}) go func() { close(stopping) - _ = manager.UnregisterTarget("", "", "test-provider", "test-ns", "main") + manager.removeWorkers([]BranchKey{key}, "test") }() <-stopping - time.Sleep(50 * time.Millisecond) // let the unregister reach Stop() + time.Sleep(50 * time.Millisecond) // let the removal reach Stop() ensured := make(chan struct{}) go func() { diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index a0edf856..e6e10f65 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -234,31 +234,6 @@ func (m *WorkerManager) SetRemoteReporter(reporter RemoteReporter) { m.remoteReporter = reporter } -// RegisterTarget ensures a worker exists for the target's (provider, branch) -// and registers the target with that worker. -// This is called by GitTarget controller when a target becomes Ready. -func (m *WorkerManager) RegisterTarget( - ctx context.Context, - targetName, targetNamespace string, - providerName, providerNamespace string, - branch, path string, -) error { - if err := m.EnsureWorker(ctx, providerName, providerNamespace, branch); err != nil { - return err - } - - m.Log.Info("GitTarget registered with branch worker", - "target", fmt.Sprintf("%s/%s", targetNamespace, targetName), - "workerKey", BranchKey{ - RepoNamespace: providerNamespace, - RepoName: providerName, - Branch: branch, - }.String(), - "path", path) - - return nil -} - // EnsureWorker ensures a worker exists for the given (provider, branch). // Worker creation/start is protected by the manager lock. func (m *WorkerManager) EnsureWorker( @@ -314,41 +289,36 @@ func (m *WorkerManager) EnsureWorker( return nil } -// UnregisterTarget removes a GitTarget from its worker. -// Destroys the worker if it was the last target using it. -// This is called by GitTarget controller when a target is deleted. -func (m *WorkerManager) UnregisterTarget( - _, _ string, - providerName, providerNamespace string, - branch string, -) error { - key := BranchKey{ - RepoNamespace: providerNamespace, - RepoName: providerName, - Branch: branch, +// removeWorkers detaches the named workers and stops them. +// +// The lock discipline is the whole reason this is one function rather than a loop at each call +// site. lifecycleMu is held across the Stops so EnsureWorker cannot start a replacement while the +// worker it replaces is still draining — they would share an on-disk clone. m.mu is held only to +// detach: queueDepthSamples reads it on the metric SDK's collection goroutine, so holding THAT +// across Stop(), which waits for the loop goroutine, stalls collection for the whole shutdown. +// +// A key that names no worker is skipped, so a caller may pass a key it is not sure about. +func (m *WorkerManager) removeWorkers(keys []BranchKey, reason string) { + if len(keys) == 0 { + return } - - // lifecycleMu is held across the Stop so EnsureWorker cannot start a replacement while this - // worker is still draining. m.mu is only held to detach: queueDepthSamples reads it on the - // scrape goroutine, so holding THAT across Stop() stalls collection instead of reporting it. m.lifecycleMu.Lock() defer m.lifecycleMu.Unlock() m.mu.Lock() - worker, exists := m.workers[key] - delete(m.workers, key) + detached := make(map[BranchKey]*BranchWorker, len(keys)) + for _, key := range keys { + if worker, exists := m.workers[key]; exists { + detached[key] = worker + delete(m.workers, key) + } + } m.mu.Unlock() - if !exists { - return nil + for key, worker := range detached { + m.Log.Info("Stopping branch worker", "key", key.String(), "reason", reason) + worker.Stop() } - - // Worker no longer tracks targets internally - always destroy worker - // since WorkerManager handles all lifecycle decisions - m.Log.Info("Unregistering target, destroying worker", "key", key.String()) - worker.Stop() - - return nil } // GetWorkerForTarget finds the worker for a target's (provider, branch). @@ -371,10 +341,22 @@ func (m *WorkerManager) GetWorkerForTarget( return worker, exists } -// ReconcileWorkers checks active GitTargets and cleans up orphaned workers. -// This ensures workers are removed when their GitTargets are deleted. +// ReconcileWorkers stops every worker no live GitTarget still needs. +// +// It is the ONLY thing that removes a worker before shutdown, and it is driven from the GitTarget +// reconcile's deleted-object path: that reconcile is what the delete watch event produces, so the +// sweep runs exactly when the set of needed workers can have shrunk. A worker that outlives its +// last GitTarget is not free — it holds a goroutine, an event queue and an on-disk clone until the +// process restarts. +// +// It decides from the API rather than from a count of registrations, which is what makes it +// correct for a branch SHARED by several GitTargets: deleting one of them leaves the others +// listed, so the worker they share is still needed and is left alone. +// +// It fails safe. A List that errors stops nothing: the alternative — treating "I could not read +// the targets" as "there are no targets" — would take down every live worker in the process. func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error { - // The List and the Stops stay outside m.mu, for the reason in UnregisterTarget. + // The List and the Stops stay outside m.mu, for the reason in removeWorkers. // Get all GitTargets var targetList configv1alpha3.GitTargetList @@ -401,30 +383,25 @@ func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error { neededWorkers[key] = true } - // Detach the orphans under m.mu; stop them after releasing it, with lifecycleMu held so no - // replacement starts while one is draining. - m.lifecycleMu.Lock() - defer m.lifecycleMu.Unlock() - - m.mu.Lock() - orphans := make(map[BranchKey]*BranchWorker) - for key, worker := range m.workers { + m.mu.RLock() + orphans := make([]BranchKey, 0, len(m.workers)) + for key := range m.workers { if !neededWorkers[key] { - orphans[key] = worker - delete(m.workers, key) + orphans = append(orphans, key) } } - remaining := len(m.workers) - m.mu.Unlock() + m.mu.RUnlock() - for key, worker := range orphans { - m.Log.Info("Cleaning up orphaned worker", "key", key.String()) - worker.Stop() - } + m.removeWorkers(orphans, "no GitTarget needs this worker any more") + + m.mu.RLock() + remaining := len(m.workers) + m.mu.RUnlock() m.Log.V(1).Info("Worker reconciliation complete", "activeWorkers", remaining, - "neededWorkers", len(neededWorkers)) + "neededWorkers", len(neededWorkers), + "stopped", len(orphans)) return nil } diff --git a/internal/git/worker_manager_test.go b/internal/git/worker_manager_test.go index b04f6571..d2d6ed19 100644 --- a/internal/git/worker_manager_test.go +++ b/internal/git/worker_manager_test.go @@ -4,6 +4,7 @@ package git import ( "context" + "errors" "path/filepath" "testing" "time" @@ -16,6 +17,7 @@ import ( clientgoscheme "k8s.io/client-go/kubernetes/scheme" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/types" @@ -97,8 +99,8 @@ func createTargetForRegister( require.NoError(t, k8sClient.Create(ctx, target)) } -// TestWorkerManagerRegisterTarget verifies worker registration. -func TestWorkerManagerRegisterTarget(t *testing.T) { +// TestEnsureWorker_CreatesAWorkerWithTheIdentityItWasAskedFor. +func TestEnsureWorker_CreatesAWorkerWithTheIdentityItWasAskedFor(t *testing.T) { scheme := setupScheme() client := fake.NewClientBuilder().WithScheme(scheme).Build() log := logr.Discard() @@ -116,10 +118,7 @@ func TestWorkerManagerRegisterTarget(t *testing.T) { createTargetForRegister(ctx, t, client, "target1", "repo1", "main", "clusters/prod") // Register first target - err := manager.RegisterTarget(ctx, - "target1", "default", - "repo1", "gitops-system", - "main", "clusters/prod") + err := manager.EnsureWorker(ctx, "repo1", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target: %v", err) } @@ -171,18 +170,12 @@ func TestWorkerManagerMultipleTargetsSameBranch(t *testing.T) { createTargetForRegister(ctx, t, client, "target-infra", "shared-repo", "main", "infra/") // Register two targets for same repo+branch, different paths - err := manager.RegisterTarget(ctx, - "target-apps", "default", - "shared-repo", "gitops-system", - "main", "apps/") + err := manager.EnsureWorker(ctx, "shared-repo", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target-apps: %v", err) } - err = manager.RegisterTarget(ctx, - "target-infra", "default", - "shared-repo", "gitops-system", - "main", "infra/") + err = manager.EnsureWorker(ctx, "shared-repo", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target-infra: %v", err) } @@ -225,18 +218,12 @@ func TestWorkerManagerDifferentBranches(t *testing.T) { createTargetForRegister(ctx, t, client, "target-dev", "repo1", "develop", "base/") // Register targets for same repo, different branches - err := manager.RegisterTarget(ctx, - "target-main", "default", - "repo1", "gitops-system", - "main", "base/") + err := manager.EnsureWorker(ctx, "repo1", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target-main: %v", err) } - err = manager.RegisterTarget(ctx, - "target-dev", "default", - "repo1", "gitops-system", - "develop", "base/") + err = manager.EnsureWorker(ctx, "repo1", "gitops-system", "develop") if err != nil { t.Fatalf("Failed to register target-dev: %v", err) } @@ -265,74 +252,73 @@ func TestWorkerManagerDifferentBranches(t *testing.T) { time.Sleep(100 * time.Millisecond) } -// TestWorkerManagerUnregisterTarget verifies target unregistration. -func TestWorkerManagerUnregisterTarget(t *testing.T) { +// targetForRegister addresses a GitTarget createTargetForRegister made, for a delete. +func targetForRegister(name string) *configv1alpha3.GitTarget { + target := &configv1alpha3.GitTarget{} + target.Name = name + target.Namespace = testTargetNamespace + return target +} + +// TestReconcileWorkers_KeepsAWorkerASiblingStillNeeds is the rule a per-target count gets wrong. +// One worker serves every GitTarget on its (provider, branch), so deleting one of them says +// nothing about the others: the sweep decides from the API, where the siblings are still listed. +func TestReconcileWorkers_KeepsAWorkerASiblingStillNeeds(t *testing.T) { scheme := setupScheme() - client := fake.NewClientBuilder().WithScheme(scheme).Build() - log := logr.Discard() + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() - manager := NewWorkerManager(client, log, BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - - go func() { - _ = manager.Start(ctx) - }() + go func() { _ = manager.Start(ctx) }() time.Sleep(100 * time.Millisecond) - createProviderWithLocalRepo(ctx, t, client, "repo1") - createTargetForRegister(ctx, t, client, "target1", "repo1", "main", "apps/") - createTargetForRegister(ctx, t, client, "target2", "repo1", "main", "infra/") - - // Register two targets - _ = manager.RegisterTarget(ctx, - "target1", "default", - "repo1", "gitops-system", - "main", "apps/") - _ = manager.RegisterTarget(ctx, - "target2", "default", - "repo1", "gitops-system", - "main", "infra/") - - // Verify worker exists - _, exists := manager.GetWorkerForTarget("repo1", "gitops-system", "main") - if !exists { - t.Fatal("Worker should exist") - } - - // Unregister first target - err := manager.UnregisterTarget("target1", "default", "repo1", "gitops-system", "main") - if err != nil { - t.Fatalf("Failed to unregister target1: %v", err) - } - // Verify worker was destroyed (WorkerManager now destroys on any unregister) - _, exists = manager.GetWorkerForTarget("repo1", "gitops-system", "main") - if exists { - t.Error("Worker should be destroyed when target unregistered") - } + createProviderWithLocalRepo(ctx, t, k8sClient, "repo1") + createTargetForRegister(ctx, t, k8sClient, "target1", "repo1", "main", "apps/") + createTargetForRegister(ctx, t, k8sClient, "target2", "repo1", "main", "infra/") + // The provider lives beside its targets, which is the namespace the sweep derives. + require.NoError(t, manager.EnsureWorker(ctx, "repo1", testTargetNamespace, "main")) - // Unregister last target - err = manager.UnregisterTarget("target2", "default", "repo1", "gitops-system", "main") - if err != nil { - t.Fatalf("Failed to unregister target2: %v", err) - } + // One of the two goes. + require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target1"))) + require.NoError(t, manager.ReconcileWorkers(ctx)) - // Verify worker was destroyed - _, exists = manager.GetWorkerForTarget("repo1", "gitops-system", "main") - if exists { - t.Error("Worker should be destroyed when last target unregistered") - } + _, exists := manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + assert.True(t, exists, "target2 still writes to this branch, so its worker must survive") - manager.mu.RLock() - finalWorkerCount := len(manager.workers) - manager.mu.RUnlock() + // And now the last one. + require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target2"))) + require.NoError(t, manager.ReconcileWorkers(ctx)) - if finalWorkerCount != 0 { - t.Errorf("Manager should have 0 workers, got %d", finalWorkerCount) - } + _, exists = manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + assert.False(t, exists, + "nothing needs this worker now, and it holds a goroutine, a queue and a clone until it stops") +} - cancel() +// TestReconcileWorkers_StopsNothingWhenTheTargetsCannotBeRead. Treating "I could not read the +// GitTargets" as "there are no GitTargets" would stop every live worker in the process, which is +// the one outcome a cleanup sweep must never produce. +func TestReconcileWorkers_StopsNothingWhenTheTargetsCannotBeRead(t *testing.T) { + scheme := setupScheme() + k8sClient := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error { + return errors.New("the cache is not having it") + }, + }).Build() + + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = manager.Start(ctx) }() time.Sleep(100 * time.Millisecond) + require.NoError(t, manager.EnsureWorker(ctx, "repo1", testTargetNamespace, "main")) + + err := manager.ReconcileWorkers(ctx) + + require.Error(t, err) + _, exists := manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + assert.True(t, exists, "a read that failed proves nothing about which workers are needed") } // TestWorkerManagerConcurrentRegistration verifies thread safety. @@ -356,13 +342,9 @@ func TestWorkerManagerConcurrentRegistration(t *testing.T) { done := make(chan bool, 10) for i := range 10 { go func(index int) { - targetName := "target" - err := manager.RegisterTarget(ctx, - targetName, "default", - "repo1", "gitops-system", - "main", "base/") + err := manager.EnsureWorker(ctx, "repo1", "gitops-system", "main") if err != nil { - t.Errorf("Failed to register target %d: %v", index, err) + t.Errorf("Failed to ensure worker %d: %v", index, err) } done <- true }(i) @@ -403,17 +385,17 @@ func TestWorkerManagerGetNonexistentWorker(t *testing.T) { } } -// TestWorkerManagerUnregisterNonexistent verifies unregistering nonexistent target is safe. -func TestWorkerManagerUnregisterNonexistent(t *testing.T) { +// TestRemoveWorkers_IgnoresAKeyThatNamesNoWorker keeps the sweep idempotent: it computes its +// orphan list outside the lifecycle lock, so a worker can have gone by the time it is removed. +func TestRemoveWorkers_IgnoresAKeyThatNamesNoWorker(t *testing.T) { scheme := setupScheme() - client := fake.NewClientBuilder().WithScheme(scheme).Build() - log := logr.Discard() + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() - manager := NewWorkerManager(client, log, BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) - // Unregister should be idempotent and not error - err := manager.UnregisterTarget("nonexistent", "default", "repo1", "gitops-system", "main") - if err != nil { - t.Errorf("Unregister nonexistent should not error: %v", err) - } + assert.NotPanics(t, func() { + manager.removeWorkers([]BranchKey{{ + RepoNamespace: "gitops-system", RepoName: "repo1", Branch: "main", + }}, "test") + }) } From 85c487a909e39216530bdddbc082087b09dcb983 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Thu, 24 Sep 2026 13:21:14 +0000 Subject: [PATCH 2/3] fix: retry a cleanup that could not read, and take the checkout with the worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps a review found in the sweep this branch wires up. **A failed read left the worker standing forever.** The sweep is the only thing that ever retires a worker, and its List can fail. Logging that and returning success meant nothing would try again: a deleted object is not re-reconciled on its own, so the worker and its clone would stand until some OTHER GitTarget happened to be deleted, which nothing guarantees. A deleted object still has a reconcile key, so the error now propagates and requeues this same pass with backoff. Every in-memory release still runs first — those cannot fail, and the one step that can must not stop them. **Stopping the goroutine was only half the leak.** Each worker keeps a clone per remote under its own (provider namespace, provider, branch) directory, and that half survives a process restart, so a cluster that creates and deletes targets accumulates checkouts until somebody notices the disk. A retired worker now takes its directory with it, after Stop rather than before — Stop waits for the loop goroutine, and until it returns the worktree can still be written. Only on the retirement path, never on shutdown: a restart wants the clones it left behind, and re-cloning every branch on every restart is the cost the on-disk cache exists to avoid. The identity guard on removeLocalState is not defensive programming. repoRootPath joins the three identity fields into a fixed prefix, so a worker with empty fields — which tests build — resolves to the ROOT of every worker's state, and deleting that would take out the live workers beside it. A test holds that line. Co-Authored-By: Claude Opus 5 --- .../gittarget_branch_target_metric_test.go | 2 +- internal/controller/gittarget_controller.go | 26 +++++-- .../gittarget_worker_cleanup_test.go | 72 ++++++++++++++++++- internal/git/branch_worker.go | 28 ++++++++ internal/git/worker_manager.go | 5 ++ internal/git/worker_manager_test.go | 49 +++++++++++++ 6 files changed, 172 insertions(+), 10 deletions(-) diff --git a/internal/controller/gittarget_branch_target_metric_test.go b/internal/controller/gittarget_branch_target_metric_test.go index de0b9b14..9fa599c4 100644 --- a/internal/controller/gittarget_branch_target_metric_test.go +++ b/internal/controller/gittarget_branch_target_metric_test.go @@ -39,7 +39,7 @@ func TestGitTargetCleanup_ForgetsTheBranchTargetMapping(t *testing.T) { require.True(t, ok, "precondition: the mapping is published") r := &GitTargetReconciler{} - r.cleanupDeletedGitTarget( + _ = r.cleanupDeletedGitTarget( context.Background(), k8stypes.NamespacedName{Namespace: "team-a", Name: "mirror"}, logf.Log, diff --git a/internal/controller/gittarget_controller.go b/internal/controller/gittarget_controller.go index 2adab12c..3dc53313 100644 --- a/internal/controller/gittarget_controller.go +++ b/internal/controller/gittarget_controller.go @@ -1147,19 +1147,31 @@ func (r *GitTargetReconciler) handleFetchError( namespacedName k8stypes.NamespacedName, ) (ctrl.Result, error) { if client.IgnoreNotFound(err) == nil { - r.cleanupDeletedGitTarget(ctx, namespacedName, log) + cleanupErr := r.cleanupDeletedGitTarget(ctx, namespacedName, log) log.Info("GitTarget not found, was likely deleted", "namespacedName", namespacedName) - return ctrl.Result{}, nil + // A deleted object still has a reconcile key, so returning the error requeues this same + // pass with backoff. That matters because the worker sweep is the ONLY thing that retires + // a worker: if its read of the GitTargets failed, swallowing the error here would leave + // the worker and its clone standing until another GitTarget somewhere is deleted, which + // nothing guarantees will ever happen. + return ctrl.Result{}, cleanupErr } log.Error(err, "unable to fetch GitTarget", "namespacedName", namespacedName) return ctrl.Result{}, err } +// cleanupDeletedGitTarget releases everything this reconciler and its data plane held for a +// GitTarget that is gone. +// +// Every step is idempotent, and they all run before the error is returned: the in-memory releases +// cannot fail, and the one step that can — the worker sweep, which reads the API — must not stop +// the others from happening. The error it returns is the caller's signal to come back, not a +// report that nothing was cleaned up. func (r *GitTargetReconciler) cleanupDeletedGitTarget( ctx context.Context, namespacedName k8stypes.NamespacedName, log logr.Logger, -) { +) error { gitDest := types.NewResourceReference(namespacedName.Name, namespacedName.Namespace) // Unconditionally, and before the EventRouter check below: the tracker is the reconciler's own // memory, so it must be released for a deleted target whether or not a data plane is wired. The @@ -1176,17 +1188,18 @@ func (r *GitTargetReconciler) cleanupDeletedGitTarget( // shrunk. The sweep decides from the API, so a branch several GitTargets share keeps its worker // while any of them is still listed. It runs ahead of the EventRouter check below because it is // the WorkerManager's business, not the data plane's. + var sweepErr error if r.WorkerManager != nil { if err := r.WorkerManager.ReconcileWorkers(ctx); err != nil { - // Not an error the reconcile can act on: the object is already gone, so there is - // nothing to requeue for. The next deletion sweeps again. log.V(1).Info("Could not sweep branch workers after a GitTarget was deleted", "error", err.Error()) + sweepErr = fmt.Errorf("sweep branch workers after %s was deleted: %w", + namespacedName.String(), err) } } if r.EventRouter == nil { - return + return sweepErr } r.EventRouter.UnregisterGitTargetEventStream(gitDest) @@ -1199,6 +1212,7 @@ func (r *GitTargetReconciler) cleanupDeletedGitTarget( } log.V(1).Info("Cleaned up in-memory state for deleted GitTarget", "gitDest", gitDest.String()) + return sweepErr } func clampIntToInt32(value int) int32 { diff --git a/internal/controller/gittarget_worker_cleanup_test.go b/internal/controller/gittarget_worker_cleanup_test.go index ea869acd..1c36ed12 100644 --- a/internal/controller/gittarget_worker_cleanup_test.go +++ b/internal/controller/gittarget_worker_cleanup_test.go @@ -4,17 +4,22 @@ package controller import ( "context" + "errors" "testing" "time" "github.com/go-logr/logr" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" k8stypes "k8s.io/apimachinery/pkg/types" clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" configbutleraiv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/git" @@ -44,7 +49,7 @@ func TestCleanupDeletedGitTarget_StopsAWorkerNothingNeedsAnyMore(t *testing.T) { require.True(t, exists) r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} - r.cleanupDeletedGitTarget(ctx, + _ = r.cleanupDeletedGitTarget(ctx, k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) _, exists = workers.GetWorkerForTarget("repo1", "shop", "main") @@ -75,7 +80,7 @@ func TestCleanupDeletedGitTarget_LeavesAWorkerAnotherGitTargetStillUses(t *testi require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} - r.cleanupDeletedGitTarget(ctx, + _ = r.cleanupDeletedGitTarget(ctx, k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") @@ -87,7 +92,68 @@ func TestCleanupDeletedGitTarget_LeavesAWorkerAnotherGitTargetStillUses(t *testi func TestCleanupDeletedGitTarget_WithoutAWorkerManager(t *testing.T) { r := &GitTargetReconciler{} assert.NotPanics(t, func() { - r.cleanupDeletedGitTarget(context.Background(), + _ = r.cleanupDeletedGitTarget(context.Background(), k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) }) } + +// TestCleanupDeletedGitTarget_RetriesWhenTheSweepCannotRead. The sweep is the only thing that ever +// retires a worker, so a read that failed must come back. A deleted object still has a reconcile +// key, so returning the error requeues this same pass; swallowing it would leave the worker and +// its clone standing until some other GitTarget happens to be deleted, which nothing guarantees. +func TestCleanupDeletedGitTarget_RetriesWhenTheSweepCannotRead(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + k8sClient := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error { + return errors.New("the cache is not having it") + }, + }).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + err := r.cleanupDeletedGitTarget(ctx, + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + + require.Error(t, err, "the caller has to come back, or nothing ever retires this worker") + _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") + assert.True(t, exists, "and the worker is still there, which is why the retry matters") +} + +// TestHandleFetchError_RequeuesAFailedCleanup is the wiring of the rule above: the reconcile must +// surface the error rather than reporting success for a pass that did not finish. +func TestHandleFetchError_RequeuesAFailedCleanup(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + k8sClient := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error { + return errors.New("the cache is not having it") + }, + }).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + _, err := r.handleFetchError(ctx, apierrors.NewNotFound( + schema.GroupResource{Resource: "gittargets"}, "checkout"), + logr.Discard(), k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}) + + assert.Error(t, err) +} diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index ea9a557f..1233e588 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "errors" "fmt" + "os" "path/filepath" "strings" "sync" @@ -347,6 +348,33 @@ func (w *BranchWorker) repoPathForRemote(remoteURL string) string { return filepath.Join(w.repoRootPath(), repoCacheKey(remoteURL)) } +// removeLocalState deletes everything this worker kept on disk: the clone of every remote it +// worked with, under its own (provider namespace, provider, branch) directory. +// +// It is called only when a worker is being retired because nothing needs it any more, never on +// shutdown — a restart wants the clones it left behind, and re-cloning every branch on every +// restart is exactly the cost the on-disk cache exists to avoid. +// +// The identity guard is not defensive programming. repoRootPath joins the three identity fields +// into a fixed prefix, so a worker with empty fields — which tests build — resolves to the ROOT of +// every worker's state, and deleting that would take out the live workers next to it. +func (w *BranchWorker) removeLocalState() { + if w.GitProviderNamespace == "" || w.GitProviderRef == "" || w.Branch == "" { + w.Log.V(1).Info("Not removing worker state: the worker has no complete identity") + return + } + // The branch directory rather than the repos/ tree inside it, so nothing is left behind to + // accumulate one empty directory per branch the operator ever mirrored. + path := filepath.Dir(w.repoRootPath()) + if err := os.RemoveAll(path); err != nil { + // Worth knowing and not worth failing for: the worker is gone either way, and what is + // left is disk rather than anything that can produce a wrong answer. + w.Log.Error(err, "Could not remove the retired worker's checkout", "path", path) + return + } + w.Log.Info("Removed the retired worker's checkout", "path", path) +} + // Start begins processing events. func (w *BranchWorker) Start(parentCtx context.Context) error { w.mu.Lock() diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index e6e10f65..ecd49c2a 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -318,6 +318,11 @@ func (m *WorkerManager) removeWorkers(keys []BranchKey, reason string) { for key, worker := range detached { m.Log.Info("Stopping branch worker", "key", key.String(), "reason", reason) worker.Stop() + // After Stop, never before: Stop waits for the loop goroutine, so until it returns the + // worktree can still be written. A worker being retired takes its clone with it — the + // goroutine was only half the leak, and the checkout is the half that survives a + // restart. + worker.removeLocalState() } } diff --git a/internal/git/worker_manager_test.go b/internal/git/worker_manager_test.go index d2d6ed19..ad4b2e59 100644 --- a/internal/git/worker_manager_test.go +++ b/internal/git/worker_manager_test.go @@ -5,6 +5,7 @@ package git import ( "context" "errors" + "os" "path/filepath" "testing" "time" @@ -399,3 +400,51 @@ func TestRemoveWorkers_IgnoresAKeyThatNamesNoWorker(t *testing.T) { }}, "test") }) } + +// TestReconcileWorkers_RemovesTheRetiredWorkersCheckout. Stopping the goroutine was only half the +// leak. Each worker keeps a clone per remote under its own (provider, branch) directory, and that +// half survives a restart, so a cluster that creates and deletes targets accumulates checkouts +// until somebody notices the disk. +func TestReconcileWorkers_RemovesTheRetiredWorkersCheckout(t *testing.T) { + scheme := setupScheme() + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() + + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = manager.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + + createProviderWithLocalRepo(ctx, t, k8sClient, "repo1") + createTargetForRegister(ctx, t, k8sClient, "target1", "repo1", "main", "apps/") + require.NoError(t, manager.EnsureWorker(ctx, "repo1", testTargetNamespace, "main")) + + worker, exists := manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + require.True(t, exists) + // Stand in for the clone a first publication would leave behind. + checkout := worker.repoPathForRemote("https://example.invalid/repo.git") + require.NoError(t, os.MkdirAll(checkout, 0o750)) + require.NoError(t, os.WriteFile(filepath.Join(checkout, "HEAD"), []byte("ref: refs/heads/main\n"), 0o600)) + t.Cleanup(func() { _ = os.RemoveAll(worker.repoRootPath()) }) + + require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target1"))) + require.NoError(t, manager.ReconcileWorkers(ctx)) + + _, err := os.Stat(checkout) + assert.True(t, os.IsNotExist(err), "the retired worker's checkout must go with it") +} + +// TestRemoveLocalState_RefusesAWorkerWithNoIdentity. repoRootPath joins the three identity fields +// into a fixed prefix, so a worker with empty fields resolves to the ROOT of every worker's state. +// Deleting that would take out the live workers beside it, which is a far worse outcome than the +// leak this reclaims. +func TestRemoveLocalState_RefusesAWorkerWithNoIdentity(t *testing.T) { + root := (&BranchWorker{Log: logr.Discard()}).repoRootPath() + require.NoError(t, os.MkdirAll(root, 0o750)) + t.Cleanup(func() { _ = os.RemoveAll(root) }) + + (&BranchWorker{Log: logr.Discard()}).removeLocalState() + + _, err := os.Stat(root) + assert.NoError(t, err, "a worker with no identity must never delete the shared root") +} From dbcb16b3694f8eb010e84fa9270d50955e6a4496 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Thu, 24 Sep 2026 14:09:51 +0000 Subject: [PATCH 3/3] fix: close three holes the bot review found in the worker sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **The sweep could stop a worker EnsureWorker had just handed out.** Extracting removeWorkers narrowed lifecycleMu to the removal, leaving the orphan SELECTION outside it — a regression against the code this branch replaced, which held the lock across both. In that window EnsureWorker finds the key present, returns "already there" without creating anything, and the sweep then stops the worker the GitTarget was just told it has, so its live events fail to route until its next reconcile. The lock spans both steps again, and removeWorkersLocked exists so the sweep does not take it twice. The window is nanoseconds wide and no amount of sleeping reaches it, so the regression test uses a seam — afterOrphanSelection, nil in production, the same device pushAtomicFn already provides for the push path. It fails without the fix and passes with it; a first attempt without the seam passed either way and was worth nothing. **A branch name with a slash put one worker's checkout inside another's.** `release/v1` is an ordinary branch name, and joining it into a path made that worker's directory a CHILD of `release`'s. Harmless while nothing deleted anything, and data loss the moment something did: retiring `release` would take the live `release/v1` worker's clone with it — which is exactly what the previous commit taught the sweep to do. A branch is now rendered as ONE component: a readable prefix for whoever reads the directory, and a hash of the raw name to keep `release/v1` and `release-v1` apart. Existing checkouts move, which costs one re-clone per branch and nothing else. **The filesystem tests wrote under the shared state root.** They are the two tests that DELETE worker state, and one of them removed /tmp's whole gitops-reverser-workers tree on cleanup — every other test's fixtures, and a developer's running operator, with it. The root is now a variable the tests point at their own directory. Co-Authored-By: Claude Opus 5 --- internal/git/branch_worker.go | 46 +++++++++++++++- .../git/branch_worker_queue_depth_test.go | 55 +++++++++++++++++++ internal/git/worker_manager.go | 35 +++++++++++- internal/git/worker_manager_test.go | 42 ++++++++++++-- 4 files changed, 167 insertions(+), 11 deletions(-) diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 1233e588..72eeef2d 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -328,18 +328,58 @@ func NewBranchWorker( } } +// workerStateRoot is where every worker's on-disk state lives. It is a variable rather than a +// constant so a test can point it at its own directory: the tests that exercise reclaiming a +// retired worker's checkout delete directories, and doing that under the shared root would take +// out the state of anything else using it, including a developer's running operator. +// +//nolint:gochecknoglobals // a test seam, like pushAtomicFn above +var workerStateRoot = filepath.Join("/tmp", "gitops-reverser-workers") + func repoCacheKey(remoteURL string) string { sum := sha256.Sum256([]byte(strings.TrimSpace(remoteURL))) return hex.EncodeToString(sum[:16]) } +// branchPathComponent renders a branch name as ONE directory component. +// +// Git branch names contain slashes — `release/v1` is ordinary — and joining one into a path makes +// its worker's directory a CHILD of the `release` worker's. That is harmless while nothing deletes +// anything, and it is a data-loss bug the moment something does: retiring `release` would take the +// live `release/v1` worker's checkout with it. +// +// The readable prefix is for whoever is looking at the directory; the hash of the raw name is what +// makes it unambiguous, including for the names sanitising would otherwise collapse together +// (`release/v1` and `release-v1`). +func branchPathComponent(branch string) string { + sum := sha256.Sum256([]byte(branch)) + digest := hex.EncodeToString(sum[:8]) + + readable := make([]rune, 0, len(branch)) + for _, r := range branch { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_': + readable = append(readable, r) + default: + readable = append(readable, '-') + } + } + if len(readable) > branchComponentReadableMax { + readable = readable[:branchComponentReadableMax] + } + return string(readable) + "-" + digest +} + +// branchComponentReadableMax caps the readable half so a long branch name cannot push the path +// past what a filesystem accepts; the digest that follows is what carries the identity. +const branchComponentReadableMax = 40 + func (w *BranchWorker) repoRootPath() string { return filepath.Join( - "/tmp", - "gitops-reverser-workers", + workerStateRoot, w.GitProviderNamespace, w.GitProviderRef, - w.Branch, + branchPathComponent(w.Branch), "repos", ) } diff --git a/internal/git/branch_worker_queue_depth_test.go b/internal/git/branch_worker_queue_depth_test.go index dc6617ea..a931088b 100644 --- a/internal/git/branch_worker_queue_depth_test.go +++ b/internal/git/branch_worker_queue_depth_test.go @@ -145,3 +145,58 @@ func TestEnsureWorker_WaitsForTheWorkerItReplacesToStop(t *testing.T) { t.Fatal("the replacement never started after its predecessor stopped") } } + +// A sweep must not stop a worker EnsureWorker has just handed to a caller. +// +// ReconcileWorkers chooses its orphans from the map and then removes them. If EnsureWorker can run +// between those two steps it finds the key present, returns "already there" without creating +// anything, and the sweep then stops the worker that GitTarget was just told it has — leaving it +// with no worker until its next reconcile, and its live events failing to route in the meantime. +// Holding lifecycleMu across the selection AND the removal is what closes it, and the seam is what +// makes that observable: without the lock the EnsureWorker below completes inside the window and +// the key ends up with no worker at all. +func TestReconcileWorkers_DoesNotStopAWorkerEnsureWorkerJustHandedOut(t *testing.T) { + withTemporaryWorkerStateRoot(t) + key := BranchKey{RepoNamespace: "test-ns", RepoName: "test-provider", Branch: "main"} + scheme := runtime.NewScheme() + require.NoError(t, configv1alpha3.AddToScheme(scheme)) + manager := &WorkerManager{ + Log: logr.Discard(), + ctx: context.Background(), + Client: fake.NewClientBuilder().WithScheme(scheme).Build(), // no GitTargets: every worker is an orphan + workers: map[BranchKey]*BranchWorker{}, + } + retiring := newMetricsTestWorker() + manager.workers[key] = retiring + + ensured := make(chan struct{}) + afterOrphanSelection = func() { + go func() { + defer close(ensured) + _ = manager.EnsureWorker(context.Background(), "test-provider", "test-ns", "main") + }() + // Long enough for an unguarded EnsureWorker to finish inside the window; a guarded one is + // blocked on lifecycleMu and finishes after the sweep instead. + time.Sleep(200 * time.Millisecond) + } + t.Cleanup(func() { afterOrphanSelection = nil }) + + require.NoError(t, manager.ReconcileWorkers(context.Background())) + + select { + case <-ensured: + case <-time.After(5 * time.Second): + t.Fatal("EnsureWorker never returned") + } + + manager.mu.RLock() + got, exists := manager.workers[key] + manager.mu.RUnlock() + t.Cleanup(func() { + if exists && got != retiring { + got.Stop() + } + }) + require.True(t, exists, "the target that asked for a worker must have one, not a slot the sweep emptied") + assert.NotSame(t, retiring, got, "and it must be a fresh one, not the worker the sweep retired") +} diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index ecd49c2a..bc6b816c 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -299,11 +299,17 @@ func (m *WorkerManager) EnsureWorker( // // A key that names no worker is skipped, so a caller may pass a key it is not sure about. func (m *WorkerManager) removeWorkers(keys []BranchKey, reason string) { + m.lifecycleMu.Lock() + defer m.lifecycleMu.Unlock() + m.removeWorkersLocked(keys, reason) +} + +// removeWorkersLocked is removeWorkers for a caller that already holds lifecycleMu, so a sweep can +// hold it across choosing the workers to retire and retiring them. +func (m *WorkerManager) removeWorkersLocked(keys []BranchKey, reason string) { if len(keys) == 0 { return } - m.lifecycleMu.Lock() - defer m.lifecycleMu.Unlock() m.mu.Lock() detached := make(map[BranchKey]*BranchWorker, len(keys)) @@ -346,6 +352,15 @@ func (m *WorkerManager) GetWorkerForTarget( return worker, exists } +// afterOrphanSelection runs between choosing the workers to retire and retiring them. It is nil in +// production and exists because that gap is the ONLY place the lock discipline in ReconcileWorkers +// can be observed: with lifecycleMu held across both steps a concurrent EnsureWorker blocks here, +// and without it that EnsureWorker hands a caller the very worker about to be stopped. The seam +// follows pushAtomicFn, which the push path already uses for the same reason. +// +//nolint:gochecknoglobals // a test seam, like pushAtomicFn +var afterOrphanSelection func() + // ReconcileWorkers stops every worker no live GitTarget still needs. // // It is the ONLY thing that removes a worker before shutdown, and it is driven from the GitTarget @@ -388,6 +403,14 @@ func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error { neededWorkers[key] = true } + // lifecycleMu is held across the SELECTION as well as the removal, and the two must not be + // separated. EnsureWorker takes this lock, finds the key present and returns "already there" + // without creating anything; if it ran between a selection made outside the lock and the + // removal, this sweep would then stop the worker that GitTarget had just been told it has, + // leaving it with no worker until its next reconcile. + m.lifecycleMu.Lock() + defer m.lifecycleMu.Unlock() + m.mu.RLock() orphans := make([]BranchKey, 0, len(m.workers)) for key := range m.workers { @@ -397,7 +420,13 @@ func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error { } m.mu.RUnlock() - m.removeWorkers(orphans, "no GitTarget needs this worker any more") + // The window the lock above closes, opened on request so a test can stand in it. Nil in + // production; see afterOrphanSelection. + if afterOrphanSelection != nil { + afterOrphanSelection() + } + + m.removeWorkersLocked(orphans, "no GitTarget needs this worker any more") m.mu.RLock() remaining := len(m.workers) diff --git a/internal/git/worker_manager_test.go b/internal/git/worker_manager_test.go index ad4b2e59..da3f760f 100644 --- a/internal/git/worker_manager_test.go +++ b/internal/git/worker_manager_test.go @@ -7,6 +7,7 @@ import ( "errors" "os" "path/filepath" + "strings" "testing" "time" @@ -406,6 +407,7 @@ func TestRemoveWorkers_IgnoresAKeyThatNamesNoWorker(t *testing.T) { // half survives a restart, so a cluster that creates and deletes targets accumulates checkouts // until somebody notices the disk. func TestReconcileWorkers_RemovesTheRetiredWorkersCheckout(t *testing.T) { + withTemporaryWorkerStateRoot(t) scheme := setupScheme() k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() @@ -425,7 +427,6 @@ func TestReconcileWorkers_RemovesTheRetiredWorkersCheckout(t *testing.T) { checkout := worker.repoPathForRemote("https://example.invalid/repo.git") require.NoError(t, os.MkdirAll(checkout, 0o750)) require.NoError(t, os.WriteFile(filepath.Join(checkout, "HEAD"), []byte("ref: refs/heads/main\n"), 0o600)) - t.Cleanup(func() { _ = os.RemoveAll(worker.repoRootPath()) }) require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target1"))) require.NoError(t, manager.ReconcileWorkers(ctx)) @@ -439,12 +440,43 @@ func TestReconcileWorkers_RemovesTheRetiredWorkersCheckout(t *testing.T) { // Deleting that would take out the live workers beside it, which is a far worse outcome than the // leak this reclaims. func TestRemoveLocalState_RefusesAWorkerWithNoIdentity(t *testing.T) { - root := (&BranchWorker{Log: logr.Discard()}).repoRootPath() - require.NoError(t, os.MkdirAll(root, 0o750)) - t.Cleanup(func() { _ = os.RemoveAll(root) }) + root := withTemporaryWorkerStateRoot(t) + require.NoError(t, os.MkdirAll(filepath.Join(root, "someone-elses-state"), 0o750)) (&BranchWorker{Log: logr.Discard()}).removeLocalState() - _, err := os.Stat(root) + _, err := os.Stat(filepath.Join(root, "someone-elses-state")) assert.NoError(t, err, "a worker with no identity must never delete the shared root") } + +// TestBranchPathComponent_KeepsOneBranchOutOfAnothersDirectory. Git branch names contain slashes, +// so joining one into a path made `release/v1`'s worker a child of `release`'s — and once a +// retired worker deletes its directory, retiring `release` would take the live worker's checkout +// with it. +func TestBranchPathComponent_KeepsOneBranchOutOfAnothersDirectory(t *testing.T) { + withTemporaryWorkerStateRoot(t) + parent := &BranchWorker{GitProviderNamespace: "shop", GitProviderRef: "repo1", Branch: "release"} + child := &BranchWorker{GitProviderNamespace: "shop", GitProviderRef: "repo1", Branch: "release/v1"} + + assert.NotContains(t, child.repoRootPath(), filepath.Dir(parent.repoRootPath())+string(filepath.Separator)+"repos", + "one branch's tree must not sit inside another's") + rel, err := filepath.Rel(filepath.Dir(parent.repoRootPath()), child.repoRootPath()) + require.NoError(t, err) + assert.True(t, strings.HasPrefix(rel, ".."), "release/v1 must not be reachable from inside release") + + // And names that sanitising alone would collapse stay apart. + lookalike := &BranchWorker{GitProviderNamespace: "shop", GitProviderRef: "repo1", Branch: "release-v1"} + assert.NotEqual(t, child.repoRootPath(), lookalike.repoRootPath()) +} + +// withTemporaryWorkerStateRoot points every worker's on-disk state at this test's own directory, +// so a test that DELETES worker state cannot reach the shared root — where another test's +// fixtures, or a developer's running operator, keep theirs. +func withTemporaryWorkerStateRoot(t *testing.T) string { + t.Helper() + previous := workerStateRoot + root := t.TempDir() + workerStateRoot = root + t.Cleanup(func() { workerStateRoot = previous }) + return root +}