worker_manager.go

// SPDX-License-Identifier: Apache-2.0

package git

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/go-logr/logr"
	k8stypes "k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/event"

	configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3"
	"github.com/ConfigButler/gitops-reverser/internal/authz"
	"github.com/ConfigButler/gitops-reverser/internal/telemetry"
	"github.com/ConfigButler/gitops-reverser/internal/types"
	"github.com/ConfigButler/gitops-reverser/internal/typeset"
)

// DefaultBranchBufferMaxBytes is the default cap on a worker's combined event
// buffer + unpushed-events memory. Operators override this via
// --branch-buffer-max-size (8Mi by default).
const DefaultBranchBufferMaxBytes int64 = 8 * 1024 * 1024

// DefaultBranchWorkerQueueDepth is the default depth of each branch worker's event
// queue. Operators override it via --branch-worker-queue-depth.
//
// It is sized so that a bounded administrative burst cannot overrun it, because a burst
// that stays under the queue depth cannot drop at ALL, whatever its arrival shape. The
// sizing input is not an event rate: a slot holds one WriteRequest, so what must fit is
// the number of concurrent write requests a burst can produce, roughly
// (concurrent writers) x (GitTargets sharing this branch worker). Note the second factor:
// workers are keyed by repository branch, so pointing another GitTarget at a branch, from any
// namespace, divides the per-target headroom of a worker already in use.
//
// 1000 replaces an earlier 100, which a realistic workload cleared and a scripted one did
// not: a 200-participant demo across two GitTargets peaked at depth 44, and then deleting
// those 203 objects unpaced -- hundreds of requests per second rather than the ballots'
// ~3/s -- dropped 20 writes. Convergence still healed the mirror, but a dropped write is a
// LIVE, ATTRIBUTED commit that never happened, so the history is missing something the
// end state cannot show.
const DefaultBranchWorkerQueueDepth = 1000

// BranchWorkerLimits bounds one branch worker's memory. The two knobs cover different
// stages of the same pipeline and neither substitutes for the other, which is why they
// travel together.
type BranchWorkerLimits struct {
	// MaxBufferBytes caps totalRetainedBytes: the open commit window plus the writes
	// committed locally and retained for replay until a push succeeds. Tripping it
	// finalizes the window early, ignoring the commit cadence. While a failed attempt waits
	// for its retry it is also the intake budget, covering queued work too. See intake.go.
	MaxBufferBytes int64

	// QueueDepth is the depth of the event queue, and it is a HARD drop boundary: the
	// enqueue is deliberately non-blocking, so a full queue refuses the item and counts
	// git_queue_refusals_total rather than stalling the watch path behind a slow remote. The
	// producer keeps it and offers it again: the watch redelivers from its unadvanced cursor.
	//
	// On a healthy branch MaxBufferBytes does NOT cover this queue: the window cap is accounted
	// only once the loop DEQUEUES an item, so whatever is still on the channel is bounded by count
	// alone and costs memory ON TOP of it. During an outage the intake gate charges queued
	// payload against MaxBufferBytes at enqueue (intake.go), in serialized bytes. The channel itself is trivial (a WorkItem is three
	// pointers); what it retains is not, because each live-event item holds a sanitized
	// object as an unstructured map, measured at roughly SIX times the bytes it serializes
	// to. Budget QueueDepth x serialized size x ~6 per SATURATED worker: ~5MiB at the
	// default depth for ~800-byte resources, but the multiplier rides the payload, so a
	// folder mirroring megabyte-sized ConfigMaps reaches gigabytes at the same depth.
	QueueDepth int
}

// withDefaults fills in the zero value of each knob, so a caller that cares about one
// need not restate the other.
func (l BranchWorkerLimits) withDefaults() BranchWorkerLimits {
	if l.MaxBufferBytes <= 0 {
		l.MaxBufferBytes = DefaultBranchBufferMaxBytes
	}
	if l.QueueDepth <= 0 {
		l.QueueDepth = DefaultBranchWorkerQueueDepth
	}
	return l
}

// WorkerManager manages BranchWorkers.
// Creates workers per (repo, branch), shared by multiple GitDestinations.
// Implements controller-runtime's Runnable interface for lifecycle management.
type WorkerManager struct {
	Client client.Client
	Log    logr.Logger

	limits             BranchWorkerLimits
	sensitiveResources types.SensitiveResourcePolicy

	mu sync.RWMutex
	// lifecycleMu serialises creating and stopping workers, so one BranchKey never has two live
	// workers: they share an on-disk clone keyed by remote URL. It is separate from mu so a
	// shutdown never blocks readers of the map, one of which is the queue-depth gauge source on
	// the scrape goroutine. Take it BEFORE mu, never the other way round.
	lifecycleMu sync.Mutex
	workers     map[BranchKey]*BranchWorker
	// commitRequests records which worker holds each CommitRequest. It outlives the map above on
	// purpose: a retired worker leaves the map before it stops, and keeps answering for the
	// requests it held. See commitRequestOwners.
	commitRequests *commitRequestOwners
	ctx            context.Context
	// mapper is the GVK->GVR resolver injected into every worker so store scans build a
	// resource-identity inventory. It is set once at startup (SetMapper) before any
	// worker is created; a nil mapper keeps workers structure-only. It is the LOCAL cluster's
	// resolver and the fallback when a GitTarget names no source cluster.
	mapper typeset.Lookup
	// clusterMapper resolves the GVK->GVR lookup for a NAMED source cluster, so a worker
	// serving a GitTarget that mirrors a remote resolves that folder against the remote's
	// registry — never a union. Set once at startup (SetClusterMapper); nil in the CLI and
	// in tests, which fall back to `mapper`.
	clusterMapper func(clusterID string) typeset.Lookup

	// sshHostKeys configures SSH host-key resolution for every worker's credential reads. Set
	// once at startup (SetSSHHostKeyConfig) before any worker is created.
	sshHostKeys SSHHostKeyConfig
	// credentialPolicy controls explicit insecure opt-ins for Git credential transports. Set
	// once at startup (SetCredentialTransportPolicy) before any worker is created.
	credentialPolicy CredentialTransportPolicy

	// pathRefusal reports a refused live write plan to the GitTarget status surface. Set
	// once at startup (SetPathRefusalReporter) before any worker is created; nil in the
	// CLI and in tests that do not assert on the status transition.
	pathRefusal PathRefusalReporter

	// layoutReporter publishes each scan's resolved folder layout to the GitTarget status
	// surface. Set once at startup (SetLayoutReporter) before any worker is created; nil in the
	// CLI and in tests that do not assert on status.placement.
	layoutReporter LayoutReporter

	// scanAcceptance publishes a read-only folder scan's verdict to the GitTarget status surface.
	// Set once at startup (SetScanAcceptanceReporter) before any worker is created; nil in the CLI
	// and in tests that do not assert on GitPathAccepted.
	scanAcceptance ScanAcceptanceReporter

	// renderFidelityGate is shared by every worker and the watch manager. It is created with the
	// manager so a target's state survives workers being recreated for the same branch.
	renderFidelityGate *RenderFidelityGate

	// remotes is what each branch was last PROVED to be at, keyed by branch and guarded by mu.
	//
	// It lives here rather than on the worker because it has to survive one: a worker replaced
	// because its GitProvider names another repository takes its own observation with it, and the
	// layer that publishes needs the old one to see that what it has published describes a
	// repository this GitTarget has left. It is dropped only when the branch itself is retired.
	remotes map[BranchKey]RemoteObservation

	// replacements are the branches whose worker was replaced and whose GitTargets have not been
	// re-established since. Guarded by mu.
	//
	// A replacement is not finished when EnsureWorker returns: the new worker has no clone, no
	// base trust and an empty queue, so every GitTarget on that branch has to be told to
	// re-anchor its streams. That delivery can fail — it lists the targets, and a list can fail —
	// and a one-shot "I replaced something" return value was then spent: later calls reported no
	// replacement, so nothing ever asked again and an idle target could leave the new repository
	// empty indefinitely. Recorded here instead, a pending recovery survives every failure until
	// something acknowledges it.
	replacements map[BranchKey]struct{}

	// publicationEvents carries the GitTargets whose worker's publication report changed. See
	// PublicationEvents.
	publicationEvents chan event.GenericEvent

	// gitBranchEvents carries the GitBranch whose worker proved something new about the remote or
	// whose publication report changed. See GitBranchEvents.
	gitBranchEvents chan event.GenericEvent

	// claims is which GitBranch holds each repository branch (branch_owner.go). The GitBranch
	// reconciler writes it; EnsureWorker and the sweep read it. Reach it through Claims, which makes
	// one for a manager not built by NewWorkerManager.
	claims     *BranchClaims
	claimsOnce sync.Once

	// initialSnapshots and resyncRecheck are the watch side's half of the applier-echo rules every
	// worker follows (apply_epoch.go). Set once at startup (SetSnapshotHooks); nil in the CLI and in
	// tests.
	initialSnapshots func(target types.ResourceReference) bool
	resyncRecheck    func(target types.ResourceReference)
	// applyBarrier lets every worker's apply windows wait for the watch (apply_barrier.go). Set once
	// at startup (SetApplyBarrier); nil in the CLI and in tests.
	applyBarrier ApplyBarrierFunc

	// applyRequester delivers every worker's apply requests. Set once at startup
	// (SetApplyRequester) before any worker is created; nil in the CLI and in tests, which switches
	// the applier claim off.
	applyRequester ApplyRequester

	// claimStatus writes what every worker publishes on its applier claims' status. Set once at
	// startup (SetApplierClaimStatusWriter); nil in the CLI and in tests, which publishes nothing.
	claimStatus ApplierClaimStatusWriter

	// applierAdmitted is the dispatcher's own check of whether a GitTarget may ask its applier claim's
	// applier (authz.ApplierAdmitted), so a worker never flushes or holds an applier for a GitTarget
	// the dispatcher would refuse. Always set by NewWorkerManager.
	applierAdmitted ApplierAdmission
}

// ApplierAdmission reports whether target may ask the applier applierClaim holds to apply.
type ApplierAdmission func(
	ctx context.Context, reader client.Reader, target *configv1alpha3.GitTarget, applierClaim configv1alpha3.ApplierClaim,
) (authz.Decision, error)

// NewWorkerManager creates a new worker manager. limits bounds every worker this manager
// creates; a zero value of either knob takes that knob's default.
func NewWorkerManager(
	client client.Client,
	log logr.Logger,
	limits BranchWorkerLimits,
	sensitiveResources types.SensitiveResourcePolicy,
) *WorkerManager {
	return &WorkerManager{
		Client:             client,
		Log:                log,
		limits:             limits.withDefaults(),
		sensitiveResources: sensitiveResources,
		workers:            make(map[BranchKey]*BranchWorker),
		commitRequests:     &commitRequestOwners{},
		remotes:            make(map[BranchKey]RemoteObservation),
		replacements:       make(map[BranchKey]struct{}),
		renderFidelityGate: NewRenderFidelityGate(),
		publicationEvents:  make(chan event.GenericEvent, publicationEventBuffer),
		gitBranchEvents:    make(chan event.GenericEvent, publicationEventBuffer),
		claims:             NewBranchClaims(),
		applierAdmitted:    authz.ApplierAdmitted,
	}
}

// Claims is the registry of which GitBranch holds each repository branch.
func (m *WorkerManager) Claims() *BranchClaims {
	m.claimsOnce.Do(func() {
		if m.claims == nil {
			m.claims = NewBranchClaims()
		}
	})
	return m.claims
}

// RenderFidelityGate returns the manager-wide target gate used by branch workers. The gate is
// safe for concurrent watch and worker access.
func (m *WorkerManager) RenderFidelityGate() *RenderFidelityGate {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.renderFidelityGate
}

// SetMapper injects the GVK->GVR resolver used by every worker's store scan. It is
// called once at startup, before any GitTarget registers a worker, so each worker
// created by EnsureWorker carries it.
func (m *WorkerManager) SetMapper(mapper typeset.Lookup) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.mapper = mapper
}

// SetClusterMapper injects the per-source-cluster GVK->GVR resolver used by every worker's
// store scan when a GitTarget names a source cluster. Like SetMapper, it is called once at
// startup before any worker is created, so each worker created by EnsureWorker carries it.
func (m *WorkerManager) SetClusterMapper(resolver func(clusterID string) typeset.Lookup) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.clusterMapper = resolver
}

// SetSnapshotHooks injects the watch side's half of docs/design/applier-echoes.md into every worker:
// settled reports whether every collection of a GitTarget has a settled snapshot, and recheck
// asks for fresh snapshots of a GitTarget whose snapshot a worker discarded as stale. Like SetMapper,
// it is called once at startup before any worker is created.
func (m *WorkerManager) SetSnapshotHooks(
	settled func(target types.ResourceReference) bool, recheck func(target types.ResourceReference),
) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.initialSnapshots, m.resyncRecheck = settled, recheck
}

// SetApplyBarrier injects the watch side's barrier, which lets an apply window wait until the watch
// delivered every write made before its apply completed (apply_barrier.go). Like SetMapper, it is
// called once at startup before any worker is created.
func (m *WorkerManager) SetApplyBarrier(barrier ApplyBarrierFunc) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.applyBarrier = barrier
}

// SetSSHHostKeyConfig injects the SSH host-key resolution config used by every worker's credential
// reads. Like SetMapper, it is called once at startup before any worker is created.
func (m *WorkerManager) SetSSHHostKeyConfig(cfg SSHHostKeyConfig) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.sshHostKeys = cfg
}

// SetCredentialTransportPolicy injects Git credential transport policy into every worker.
func (m *WorkerManager) SetCredentialTransportPolicy(policy CredentialTransportPolicy) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.credentialPolicy = policy
}

// SetPathRefusalReporter injects the hook every worker calls when a live write plan is
// refused, so the refusal reaches GitTarget status instead of being logged and dropped. Like
// SetMapper, it is called once at startup before any worker is created.
func (m *WorkerManager) SetPathRefusalReporter(reporter PathRefusalReporter) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.pathRefusal = reporter
}

// SetLayoutReporter injects the hook every worker calls after a scan resolves a target's
// folder layout, so status.placement and LayoutResolved reflect the folder rather than being
// computed and dropped. Like SetPathRefusalReporter, it is called once at startup before any
// worker is created.
func (m *WorkerManager) SetLayoutReporter(reporter LayoutReporter) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.layoutReporter = reporter
}

// SetScanAcceptanceReporter injects the hook the refresher calls after it re-reads a target's
// folder, so a structural refusal that arrived in Git surfaces before anything tries to write it.
// Like SetLayoutReporter, it is called once at startup before any worker is created.
func (m *WorkerManager) SetScanAcceptanceReporter(reporter ScanAcceptanceReporter) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.scanAcceptance = reporter
}

// SetApplyRequester injects what delivers every worker's apply requests. Like SetMapper, it is called
// once at startup before any worker is created.
func (m *WorkerManager) SetApplyRequester(requester ApplyRequester) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.applyRequester = requester
}

// SetApplierClaimStatusWriter injects what writes every worker's applier claim status. Like
// SetApplyRequester, it is called once at startup before any worker is created.
func (m *WorkerManager) SetApplierClaimStatusWriter(writer ApplierClaimStatusWriter) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.claimStatus = writer
}

// recordRemoteObservation is the delivery point for one branch's confirmed observation. Every
// worker is given it bound to its own key; it is called on the worker's loop goroutine.
func (m *WorkerManager) recordRemoteObservation(key BranchKey, observed RemoteObservation) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.remotes == nil {
		m.remotes = map[BranchKey]RemoteObservation{}
	}
	m.remotes[key] = observed
	if worker, live := m.workers[key]; live {
		m.notifyGitBranchLocked(worker.GitBranch)
	}
}

// RemoteForBranch is what this branch was last proved to be at, and whether anything has looked at
// all. It opens no connection and reads no repository.
//
// It answers for the BRANCH, which is the whole of the delivery half: two GitTargets writing
// different folders of one branch share a worker, so they share its observation and the moment it
// was proved. Absent means nothing has looked yet, which is different from a present observation
// with no revision: that one means the branch is not on the remote.
func (m *WorkerManager) RemoteForBranch(key BranchKey) (RemoteObservation, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	observed, known := m.remotes[key]
	return observed, known
}

// WorkerRequest names the branch worker a GitTarget needs: the GitBranch holding the branch, the
// GitProvider it writes with, and the repository that provider names.
type WorkerRequest struct {
	// GitBranch is the claim holder the worker writes for.
	GitBranch k8stypes.NamespacedName
	// Provider is the GitBranch's GitProvider, its namespace resolved: a GitBranch may name one in
	// another namespace, whose credential the worker then pushes with.
	Provider k8stypes.NamespacedName
	// Branch is the GitBranch's spec.branch.
	Branch string
	// Repo is the repository the GitProvider names now.
	Repo RepoIdentity
	// ParentBranch is the GitBranch's spec.parentBranch.
	ParentBranch string
}

// EnsureWorker ensures a worker exists for the repository branch req names, keyed by that branch
// (BranchKey), and that it is the worker for the repository the GitProvider names NOW. Only the
// GitBranch holding the branch's claim (branch_owner.go) gets one: any other is refused
// (ErrBranchOwnedElsewhere), and so is a request while another GitBranch's worker is still live on
// the branch, which the sweep retires.
//
// req.Repo is the caller's answer to "which repository is this"; the reconcile has already read the
// GitProvider, so it costs no API call. A slot holding a worker for a DIFFERENT repository is not
// corrected field by field — the clone, the base trust, the observation and the retained writes are
// all statements about the old one — so that worker is stopped, its checkout reclaimed, and a fresh
// one takes the slot.
//
// A replacement is not complete when it returns: the new worker has no clone and no queue, so
// every GitTarget on that branch has to re-establish its folder against the new repository. That
// is recorded as a PENDING RECOVERY rather than returned, because arranging it is the caller's job
// and the caller can fail. See ReplacementPending; the delivery is the GitTarget reconcile's
// worker wiring gate.
//
// req.ParentBranch is immutable on a GitBranch, so a live worker sees a different one only when its
// GitBranch was replaced before the worker sweep ran. That is not a replacement: the worker keeps
// its clone, and SetParentBranch raises the parent generation, so nothing planned under the old
// parent is pushed and retained writes are rebuilt on the new one.
//
// Worker creation/start is protected by the manager lock.
func (m *WorkerManager) EnsureWorker(_ context.Context, req WorkerRequest) error {
	// Held across the whole check-and-create so a replacement cannot start while the worker it
	// replaces is still stopping; they would share a clone.
	m.lifecycleMu.Lock()
	defer m.lifecycleMu.Unlock()

	repo := req.Repo
	key := BranchKeyFor(repo.URL, req.Branch)
	if key == (BranchKey{}) {
		return fmt.Errorf("GitProvider %s names no repository", req.Provider)
	}
	holder, err := m.refuseUnclaimed(key, req.GitBranch)
	if err != nil {
		return err
	}

	m.mu.RLock()
	existing, exists := m.workers[key]
	previous, hadPrevious := m.workerServingLocked(req.GitBranch)
	m.mu.RUnlock()

	if exists && existing.GitBranch != req.GitBranch {
		return fmt.Errorf("%w: %s, by GitBranch %s", ErrBranchOwnedElsewhere, key, existing.GitBranch)
	}
	if !exists && hadPrevious && previous != key {
		// This GitBranch's worker writes another repository: its GitProvider was recreated under
		// its name with another URL. The new repository is a new key, and the old worker's clone,
		// base trust and retained writes are about the old repository.
		m.Log.Info("GitProvider now names a different repository; replacing the branch worker",
			"key", key.String(), "was", previous.String())
		m.removeWorkersLocked([]BranchKey{previous}, "the GitProvider names a different repository")
		m.noteReplacement(key)
		m.carryRemote(previous, key)
	}
	if exists {
		if existing.repo == repo {
			if existing.SetParentBranch(req.ParentBranch) {
				m.Log.Info("Parent branch changed; the worker re-reads the remote on its next cycle",
					"key", key.String(), "parentBranch", req.ParentBranch)
			}
			return nil
		}
		// Both identities, at default verbosity: a comparison that is not stable across steady
		// reconciles is a restart loop, and this line is what makes one visible.
		m.Log.Info("GitProvider now names a different repository; replacing the branch worker",
			"key", key.String(),
			"was", existing.repo.String(),
			"now", repo.String())
		// lifecycleMu is already held, so the detach-and-stop must not take it again.
		m.removeWorkersLocked([]BranchKey{key}, "the GitProvider names a different repository")
		m.noteReplacement(key)
	}

	m.mu.Lock()
	defer m.mu.Unlock()

	if _, live := m.workers[key]; !live {
		return m.startWorkerLocked(key, req, holder.UID)
	}

	return nil
}

// refuseUnclaimed is EnsureWorker's claim check: only the GitBranch holding key gets its worker.
// It returns that holder.
func (m *WorkerManager) refuseUnclaimed(key BranchKey, gitBranch k8stypes.NamespacedName) (BranchClaimant, error) {
	holder, held := m.Claims().Holder(key)
	switch {
	case !held:
		return BranchClaimant{}, fmt.Errorf("%w: %s, and no GitBranch holds it yet", ErrBranchOwnedElsewhere, key)
	case holder.GitBranch != gitBranch:
		return BranchClaimant{}, fmt.Errorf("%w: %s, held by GitBranch %s",
			ErrBranchOwnedElsewhere, key, holder.GitBranch)
	}
	return holder, nil
}

// startWorkerLocked creates and starts the worker for key, writing through provider, and records
// it. gitBranchUID is the claim holder's UID, read from the claim rather than the request because
// the claim is what EnsureWorker just checked; it names the worker's checkout. Called with
// lifecycleMu and mu held.
func (m *WorkerManager) startWorkerLocked(key BranchKey, req WorkerRequest, gitBranchUID k8stypes.UID) error {
	provider, repo, parentBranch := req.Provider, req.Repo, req.ParentBranch
	providerName, providerNamespace, branch := provider.Name, provider.Namespace, key.Branch
	m.Log.Info("Creating new branch worker", "key", key.String(), "repository", repo.String())
	worker := NewBranchWorker(
		m.Client,
		m.Log.WithName("branch-worker"),
		providerName,
		providerNamespace,
		branch,
		repo,
		newContentWriter(m.sensitiveResources),
		m.limits,
	)
	// Inject the resolver before Start: the field is read only by the event-loop
	// goroutine Start spawns, so setting it here (under m.mu, before that goroutine
	// exists) is race-free.
	worker.GitBranch = req.GitBranch
	worker.GitBranchUID = gitBranchUID
	worker.SetParentBranch(parentBranch)
	worker.mapper = m.mapper
	worker.clusterMapper = m.clusterMapper
	worker.sshHostKeys = m.sshHostKeys
	worker.credentialPolicy = m.credentialPolicy
	worker.pathRefusal = m.pathRefusal
	worker.layoutReporter = m.layoutReporter
	worker.remoteReporter = func(observed RemoteObservation) { m.recordRemoteObservation(key, observed) }
	managerCtx := m.ctx // read under m.mu; the worker's loop calls the reporter later
	gitBranch := req.GitBranch
	worker.publicationReporter = func() { go m.notifyPublication(managerCtx, gitBranch) }
	worker.scanAcceptance = m.scanAcceptance
	worker.applyRequester = m.applyRequester
	worker.claimStatus = m.claimStatus
	worker.initialSnapshots = m.initialSnapshots
	worker.resyncRecheck = m.resyncRecheck
	worker.applyBarrier = m.applyBarrier
	worker.replacementPending = func() bool { return m.ReplacementPending(key) }
	worker.applierAdmitted = m.applierAdmitted
	worker.renderFidelityGate = m.renderFidelityGate
	worker.crOwners = m.commitRequests

	if err := worker.Start(m.ctx); err != nil {
		return fmt.Errorf("failed to start worker for %s: %w", key.String(), err)
	}

	m.workers[key] = worker
	return nil
}

// ErrBranchOwnedElsewhere is EnsureWorker's refusal to start a worker for a GitBranch that does not
// hold its repository branch (branch_owner.go). The GitTarget reconciler checks the claim first;
// this holds while a claim passes to another GitBranch, until the sweep has retired the previous
// holder's worker and the new holder's next reconcile starts its own.
var ErrBranchOwnedElsewhere = errors.New("another GitBranch holds this repository branch")

// workerServingLocked is the key of the worker that writes for gitBranch, if one is live. Called
// with m.mu held. There are few workers, so it looks at each.
func (m *WorkerManager) workerServingLocked(gitBranch k8stypes.NamespacedName) (BranchKey, bool) {
	for key, worker := range m.workers {
		if worker.GitBranch == gitBranch {
			return key, true
		}
	}
	return BranchKey{}, false
}

// notifyGitBranchLocked wakes the GitBranch reconciler for gitBranch. It never blocks: a full buffer
// drops the event, and the GitBranch's own requeue reads the worker later. Called with m.mu held or
// from the worker's reporter goroutine; it touches only the channel.
func (m *WorkerManager) notifyGitBranchLocked(gitBranch k8stypes.NamespacedName) {
	if m.gitBranchEvents == nil || gitBranch.Name == "" {
		return
	}
	object := &configv1alpha3.GitBranch{}
	object.Name, object.Namespace = gitBranch.Name, gitBranch.Namespace
	select {
	case m.gitBranchEvents <- event.GenericEvent{Object: object}:
	default:
	}
}

// WakeGitBranch asks the GitBranch reconciler to look at gitBranch now: a branch claim moved to it.
func (m *WorkerManager) WakeGitBranch(gitBranch k8stypes.NamespacedName) {
	m.notifyGitBranchLocked(gitBranch)
}

// GitBranchEvents carries a GitBranch whose branch worker proved something new about the remote or
// whose publication report changed, so the GitBranch controller publishes status.remote within one
// reconcile instead of at its next requeue. Nil for a manager not built by NewWorkerManager.
func (m *WorkerManager) GitBranchEvents() <-chan event.GenericEvent { return m.gitBranchEvents }

// carryRemote moves the last observation of a replaced worker's branch to the key of the worker
// replacing it. It is stamped with the old repository, which is how the GitTarget's status sees
// that what it published describes a repository it has left, and takes it back. Called with
// lifecycleMu held; it takes mu itself.
func (m *WorkerManager) carryRemote(from, to BranchKey) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if observed, known := m.remotes[from]; known {
		if _, proved := m.remotes[to]; !proved {
			m.remotes[to] = observed
		}
		delete(m.remotes, from)
	}
}

// noteReplacement records that a branch's GitTargets have to be re-established. It is called with
// lifecycleMu held and takes mu itself, which is the lock order everything here uses.
func (m *WorkerManager) noteReplacement(key BranchKey) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.replacements == nil {
		m.replacements = map[BranchKey]struct{}{}
	}
	m.replacements[key] = struct{}{}
}

// ReplacementPending reports whether this branch's GitTargets still have to be re-established
// after their worker was replaced.
//
// It answers for the BRANCH rather than for the caller, so whichever GitTarget reconciles next
// picks the recovery up — including the siblings, which have no reason of their own to notice
// that the repository changed.
func (m *WorkerManager) ReplacementPending(key BranchKey) bool {
	m.mu.RLock()
	defer m.mu.RUnlock()
	_, pending := m.replacements[key]
	return pending
}

// AcknowledgeReplacement clears a pending recovery, and must be called only once the targets on
// the branch have actually been told. Anything less leaves the request standing, which is the
// whole point of keeping it here: a failed delivery is retried by the next reconcile of any
// GitTarget on the branch, and by the periodic sweep that follows them.
func (m *WorkerManager) AcknowledgeReplacement(key BranchKey) {
	m.mu.Lock()
	defer m.mu.Unlock()
	delete(m.replacements, key)
}

// 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) {
	m.lifecycleMu.Lock()
	defer m.lifecycleMu.Unlock()
	m.removeWorkersLocked(keys, reason)
}

// removeWorkersLocked is removeWorkers for a caller that already holds lifecycleMu: a sweep holding
// it across choosing the workers to retire and retiring them, and EnsureWorker holding it across
// its whole check-and-create, where a second acquisition would deadlock.
func (m *WorkerManager) removeWorkersLocked(keys []BranchKey, reason string) {
	if len(keys) == 0 {
		return
	}

	m.mu.Lock()
	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()

	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()
	}
}

// RetireGitBranch stops the worker writing for gitBranch, if one is live, and returns once it has
// stopped. Stopping closes the worker's intake, pushes or refuses every pending write and settles
// every CommitRequest it holds, so a GitBranch being deleted calls this before its finalizer goes.
// What the manager kept about the branch goes with it: the next holder starts from the remote.
func (m *WorkerManager) RetireGitBranch(gitBranch k8stypes.NamespacedName) {
	m.lifecycleMu.Lock()
	defer m.lifecycleMu.Unlock()
	m.mu.RLock()
	key, live := m.workerServingLocked(gitBranch)
	m.mu.RUnlock()
	if !live {
		return
	}
	m.removeWorkersLocked([]BranchKey{key}, "its GitBranch is being deleted")
	m.mu.Lock()
	delete(m.replacements, key)
	delete(m.remotes, key)
	m.mu.Unlock()
}

// WorkerForGitBranch finds the worker writing for gitBranch: the worker of the repository branch
// that GitBranch holds. A GitTarget reaches its worker through its spec.gitBranchRef, so this is
// the lookup the event router and the GitTarget reconciler use, with no read on the event path.
func (m *WorkerManager) WorkerForGitBranch(gitBranch k8stypes.NamespacedName) (*BranchWorker, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	key, exists := m.workerServingLocked(gitBranch)
	if !exists {
		return nil, false
	}
	return m.workers[key], true
}

// WorkerForGitBranchAt is WorkerForGitBranch, but only for the worker of key: while a GitProvider is
// recreated for another repository, the worker of the repository it left still serves the GitBranch
// until the GitTarget reconcile replaces it, and what it observed is not about key.
func (m *WorkerManager) WorkerForGitBranchAt(gitBranch k8stypes.NamespacedName, key BranchKey) (*BranchWorker, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	worker, exists := m.workers[key]
	if !exists || worker.GitBranch != gitBranch {
		return nil, false
	}
	return worker, true
}

// WorkerForTarget is WorkerForGitBranch for the GitBranch a GitTarget references.
func (m *WorkerManager) WorkerForTarget(target *configv1alpha3.GitTarget) (*BranchWorker, bool) {
	return m.WorkerForGitBranch(target.GitBranchKey())
}

// CommitRequestOwner returns the worker that holds a CommitRequest: the one that accepted its
// attach and has not given it back. It is found whatever has happened to the request's GitTarget
// since, including a worker retired and still finishing its last push, so a caller that finds no
// owner knows that no worker can still commit the request.
func (m *WorkerManager) CommitRequestOwner(namespace, name, uid string) (*BranchWorker, bool) {
	return m.commitRequests.owner(commitRequestID{Namespace: namespace, Name: name, UID: uid})
}

// 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, then deletes the checkouts
// no GitBranch names (reclaimOrphanedCheckoutsLocked).
//
// 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 {
	// lifecycleMu is held across the whole decision — the List, the selection and the removal —
	// and the three must not be separated.
	//
	// EnsureWorker takes this lock, finds the key present and returns "already there" without
	// creating anything, so a selection made outside it would let this sweep stop a worker a
	// GitTarget had just been told it has. The LIST is inside for the same reason read the other
	// way round: a target created after the snapshot is taken is absent from it, and its worker —
	// created under this lock moments later — then looks like an orphan and is stopped with its
	// queue and its checkout. The client is cached, so the read costs no round trip.
	//
	// m.mu is still taken only for the map itself, and the Stops still run outside it, for the
	// reason in removeWorkers.
	m.lifecycleMu.Lock()
	defer m.lifecycleMu.Unlock()

	// Get all GitTargets
	var targetList configv1alpha3.GitTargetList
	if err := m.Client.List(ctx, &targetList); err != nil {
		return fmt.Errorf("failed to list GitTargets: %w", err)
	}

	needed := workersNeeded(targetList.Items, m.Claims())

	m.mu.RLock()
	orphans := make([]BranchKey, 0, len(m.workers))
	for key, worker := range m.workers {
		if gitBranch, kept := needed[key]; !kept || gitBranch != worker.GitBranch {
			orphans = append(orphans, key)
		}
	}
	m.mu.RUnlock()

	// 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")
	// A branch no GitTarget needs has nothing left to re-establish and nothing left to report: a
	// recovery or an observation held for it would outlive every object it was about.
	m.mu.Lock()
	for _, key := range orphans {
		delete(m.replacements, key)
		delete(m.remotes, key)
	}
	m.mu.Unlock()

	m.mu.RLock()
	remaining := len(m.workers)
	m.mu.RUnlock()

	m.Log.V(1).Info("Worker reconciliation complete",
		"activeWorkers", remaining,
		"neededWorkers", len(needed),
		"stopped", len(orphans))

	return m.reclaimOrphanedCheckoutsLocked(ctx)
}

// reclaimOrphanedCheckoutsLocked deletes every directory under the worker state root that no
// GitBranch names: one whose GitBranch was deleted while no worker ran for it, which no retiring
// worker was there to remove. That happens across a restart that keeps the disk, such as a
// container restart inside its pod, and for a GitBranch deleted after its last GitTarget.
//
// A directory is kept while a live worker uses it or a GitBranch with its UID exists, so a
// GitBranch whose worker has not started yet keeps its checkout for when it does. Called with
// lifecycleMu held, so no worker starts while the directories are read: one that starts later
// creates its directory after the sweep. It fails safe like the worker sweep: a List that errors
// deletes nothing.
func (m *WorkerManager) reclaimOrphanedCheckoutsLocked(ctx context.Context) error {
	var branches configv1alpha3.GitBranchList
	if err := m.Client.List(ctx, &branches); err != nil {
		return fmt.Errorf("failed to list GitBranches: %w", err)
	}
	kept := make(map[string]bool, len(branches.Items))
	for i := range branches.Items {
		kept[string(branches.Items[i].UID)] = true
	}
	for _, worker := range m.liveWorkers() {
		kept[worker.stateComponent()] = true
	}

	entries, err := os.ReadDir(workerStateRoot)
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			return nil
		}
		return fmt.Errorf("failed to read the worker state root: %w", err)
	}
	for _, entry := range entries {
		if kept[entry.Name()] {
			continue
		}
		path := filepath.Join(workerStateRoot, entry.Name())
		if err := os.RemoveAll(path); err != nil {
			m.Log.Error(err, "Could not remove an orphaned checkout", "path", path)
			continue
		}
		m.Log.Info("Removed an orphaned checkout: no GitBranch names it", "path", path)
	}
	return nil
}

// Start implements manager.Runnable interface.
// This is called by controller-runtime when the manager starts.
func (m *WorkerManager) Start(ctx context.Context) error {
	// Publish the context under m.mu: EnsureWorker reads m.ctx under the same lock,
	// so guarding the write makes that read race-free regardless of call ordering.
	m.mu.Lock()
	m.ctx = ctx
	m.mu.Unlock()
	telemetry.SetGaugeSource(telemetry.GaugeGitQueueDepth, m.queueDepthSamples)
	m.setPublicationGaugeSources()
	m.Log.Info("WorkerManager started")

	m.sweepPeriodically(ctx)

	// Clear the source before the workers go, so the callback cannot outlive them.
	telemetry.SetGaugeSource(telemetry.GaugeGitQueueDepth, nil)
	m.Log.Info("WorkerManager shutting down")
	m.lifecycleMu.Lock()
	defer m.lifecycleMu.Unlock()

	m.mu.Lock()
	workers := m.workers
	m.workers = make(map[BranchKey]*BranchWorker)
	m.mu.Unlock()

	// Stop all workers gracefully
	for key, worker := range workers {
		m.Log.Info("Stopping worker for shutdown", "key", key.String())
		worker.Stop()
	}
	m.Log.Info("WorkerManager stopped")
	return nil
}

// WorkerSweepInterval is the floor under the orphan sweep. It is a FLOOR and not the mechanism: a
// deleted GitTarget's reconcile sweeps immediately, and this catches what that path cannot see.
const WorkerSweepInterval = time.Minute

// sweepPeriodically runs the orphan sweep until the context ends.
//
// The delete-triggered sweep is prompt but not complete, because it runs only where a reconcile
// READ NotFound. Delete a GitTarget and recreate it under the same name on another branch before
// that reconcile runs, and the controller sees the successor, wires its worker, and never learns
// that the predecessor existed: the old branch's worker, its goroutine and its clone are then held
// until the process restarts. Kubernetes reconciliation cannot be built on observing every
// intermediate state, so the sweep needs an applier claim that does not depend on seeing the delete.
//
// It is a separate goroutine rather than a step in the GitTarget reconcile on purpose: the sweep
// takes lifecycleMu and holds it across Stop, and GitTarget reconciles are serialized, so a slow
// stop on the reconcile path would stall every target's reconcile behind it.
func (m *WorkerManager) sweepPeriodically(ctx context.Context) {
	ticker := time.NewTicker(WorkerSweepInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := m.ReconcileWorkers(ctx); err != nil {
				// Failing safe: a List that errors stops nothing, so the only cost is that the
				// sweep is late.
				m.Log.V(1).Info("Periodic worker sweep could not read the cluster",
					"error", err.Error())
			}
		}
	}
}

// queueDepthSamples is the git_queue_depth source: one sample per live worker, read when
// Prometheus scrapes.
//
// It takes m.mu only, and only to copy the worker pointers out; each worker's own depth is two
// atomic loads. That matters because the callback runs inside the metric SDK's collection path: a
// source that waited on the lock a wedged worker holds across its slow work would reintroduce the
// staleness the observable gauge exists to remove.
func (m *WorkerManager) queueDepthSamples() []telemetry.GaugeSample {
	workers := m.liveWorkers()
	samples := make([]telemetry.GaugeSample, 0, len(workers))
	for _, worker := range workers {
		samples = append(samples, telemetry.GaugeSample{
			Value: worker.queueDepth(),
			Attrs: worker.gitBranchAttrs(),
		})
	}
	return samples
}

// liveWorkers copies the worker pointers out under m.mu, for a gauge source to read each worker's
// atomics without holding the lock.
func (m *WorkerManager) liveWorkers() []*BranchWorker {
	m.mu.RLock()
	defer m.mu.RUnlock()
	workers := make([]*BranchWorker, 0, len(m.workers))
	for _, worker := range m.workers {
		workers = append(workers, worker)
	}
	return workers
}

// NeedLeaderElection ensures only the elected leader manages workers.
// This prevents multiple pods from managing the same workers.
func (m *WorkerManager) NeedLeaderElection() bool {
	return true
}

// NotifyPush tells the worker for a provider's branch that the Git host reported a push to it, and
// wakes its GitBranch, so status.lastPushNoticeAt records the receipt even when the look it causes
// cannot read the remote. It reports whether such a worker exists. after is the commit the delivery
// named; it is logged and never trusted: the worker reads where the branch is from the remote itself.
func (m *WorkerManager) NotifyPush(providerNamespace, providerName, branch, after string) bool {
	provider := k8stypes.NamespacedName{Namespace: providerNamespace, Name: providerName}
	m.mu.RLock()
	var found *BranchWorker
	for key, worker := range m.workers {
		if worker.provider() == provider && key.Branch == branch {
			found = worker
			break
		}
	}
	m.mu.RUnlock()
	if found == nil {
		return false
	}
	found.EnqueuePushNotification(after)
	m.notifyGitBranchLocked(found.GitBranch)
	return true
}

// workersNeeded is the worker each repository branch needs, by the GitBranch it writes for: the
// holder of a branch that at least one live GitTarget references. A held branch nothing references
// needs no worker, and one whose GitTargets are all gone has its worker retired.
func workersNeeded(targets []configv1alpha3.GitTarget, claims *BranchClaims) map[BranchKey]k8stypes.NamespacedName {
	needed := map[BranchKey]k8stypes.NamespacedName{}
	for i := range targets {
		target := &targets[i]
		if !target.DeletionTimestamp.IsZero() {
			continue
		}
		gitBranch := target.GitBranchKey()
		if gitBranch.Name == "" {
			continue
		}
		if key, _, held := claims.HeldBy(gitBranch); held {
			needed[key] = gitBranch
		}
	}
	return needed
}