fix: stop leaking a branch worker when nothing needs it any more

.coverage-baseline

@@ -1 +1 @@
-78.7
+78.9

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",

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,23 +1141,37 @@ 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)
+		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
@@ -1169,8 +1183,23 @@ 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.
+	var sweepErr error
+	if r.WorkerManager != nil {
+		if err := r.WorkerManager.ReconcileWorkers(ctx); err != nil {
+			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)
@@ -1183,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 {

internal/controller/gittarget_worker_cleanup_test.go

@@ -0,0 +1,159 @@
+// SPDX-License-Identifier: Apache-2.0
+
+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"
+	"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())
+	})
+}
+
+// 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)
+}

internal/git/branch_worker.go

@@ -8,6 +8,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"os"
 	"path/filepath"
 	"strings"
 	"sync"
@@ -327,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",
 	)
 }
@@ -347,6 +388,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()