feat(git): make the branch worker queue depth configurable, and raise it

charts/gitops-reverser/README.md

@@ -177,6 +177,8 @@ nodeSelector:
 | `servers.audit.tls.secretNameOverride` | Override Secret name for audit TLS cert/key | `<release>-audit-server-cert` |
 | `controllerManager.additionalSensitiveResources` | Extra Secret-shaped resource types encrypted as `resource` or `group/resource` | `[]` |
 | `controllerManager.allowInsecureGitHTTP` | Permit credentials with `http://` GitProvider URLs for trusted in-cluster development Git servers | `false` |
+| `controllerManager.branchWorkerQueueDepth` | Event queue depth per branch worker. A full queue drops the write (`git_queue_drops_total`) instead of stalling the watch path, so the live attributed commit is lost even though convergence heals the mirror. Costs queue depth × payload of pod memory, which `branchBufferMaxSize` does **not** cover | `1000` |
+| `controllerManager.branchBufferMaxSize` | Maximum in-memory event buffer per branch worker. Bounds the open commit window plus the writes retained for replay until a push succeeds | `8Mi` |
 | `auditService.type` | Service type for the dedicated audit Service | `NodePort` |
 | `auditService.nodePort` | Fixed NodePort for the audit Service when `auditService.type=NodePort` | `30444` |
 | `auditService.clusterIP` | Optional fixed ClusterIP for the dedicated audit Service | `""` |

charts/gitops-reverser/templates/deployment.yaml

@@ -105,6 +105,8 @@ spec:
             {{- if .Values.controllerManager.allowInsecureGitHTTP }}
             - --allow-insecure-git-http
             {{- end }}
+            - --branch-worker-queue-depth={{ int .Values.controllerManager.branchWorkerQueueDepth }}
+            - --branch-buffer-max-size={{ .Values.controllerManager.branchBufferMaxSize }}
             {{- if .Values.logging.level }}
             - --zap-log-level={{ .Values.logging.level }}
             {{- end }}

charts/gitops-reverser/values.schema.json

@@ -103,6 +103,15 @@
         "allowInsecureGitHTTP": {
           "type": "boolean",
           "description": "Permit credentials with http:// GitProvider URLs for trusted in-cluster development Git servers."
+        },
+        "branchWorkerQueueDepth": {
+          "type": "integer",
+          "minimum": 1,
+          "description": "Event queue depth per branch worker. A full queue drops the write (git_queue_drops_total) rather than stalling the watch path, losing that live attributed commit. Costs queue depth x payload of pod memory, which branchBufferMaxSize does not cover."
+        },
+        "branchBufferMaxSize": {
+          "type": "string",
+          "description": "Maximum in-memory event buffer per branch worker, as a Kubernetes resource quantity (e.g. 8Mi). Bounds the open commit window plus the writes retained for replay until a push succeeds."
         }
       }
     },

charts/gitops-reverser/values.yaml

@@ -64,6 +64,28 @@ controllerManager:
   # Permit credentials with http:// GitProvider URLs. This is only for trusted
   # in-cluster development Git servers; production remotes should use HTTPS or SSH.
   allowInsecureGitHTTP: false
+  # Event queue depth per branch worker. One branch worker is shared by every GitTarget
+  # writing to one (GitProvider, branch), so pointing another GitTarget at a branch
+  # divides the headroom of a worker already in use.
+  #
+  # A full queue DROPS the write and counts git_queue_drops_total, rather than stalling
+  # the watch path behind a slow remote. Convergence heals the mirror afterwards, but the
+  # live attributed commit for a dropped write never happens — so raise this if
+  # git_queue_drops_total is ever non-zero. Size it so a bounded burst cannot overrun it:
+  # roughly (concurrent writers) x (GitTargets sharing the branch worker). Note that an
+  # administrative burst (an unpaced `kubectl delete` of many objects) is far spikier than
+  # the steady-state traffic the depth gauge shows you.
+  #
+  # Costs pod memory ON TOP of branchBufferMaxSize, which does NOT cover this queue: that
+  # cap is accounted only after an item leaves it. Budget queue depth x serialized size
+  # x ~6 (a queued object is held as an unstructured map), per SATURATED worker — about
+  # 5Mi at this default for ~800-byte resources, but the multiplier is on the payload, so
+  # size it against your own if you mirror megabyte-sized ConfigMaps or Secrets.
+  branchWorkerQueueDepth: 1000
+  # Maximum in-memory event buffer per branch worker, as a Kubernetes resource quantity.
+  # Bounds the open commit window plus the writes retained for replay until a push
+  # succeeds; tripping it finalizes the window early, ignoring the commit cadence.
+  branchBufferMaxSize: 8Mi
 
 # cert-manager issuer shared by every certificate the chart mints. One self-signed CA
 # backs them all, so the issuer is genuinely cross-cutting and lives here; each server

cmd/main.go

@@ -131,7 +131,7 @@ func main() {
 	workerManager := git.NewWorkerManager(
 		mgr.GetClient(),
 		ctrl.Log.WithName("worker-manager"),
-		cfg.branchBufferMaxBytes,
+		cfg.branchWorkerLimits,
 		cfg.sensitiveResources,
 	)
 	workerManager.SetSSHHostKeyConfig(cfg.sshHostKeys)
@@ -451,11 +451,11 @@ type appConfig struct {
 	// check: the in-memory fact transport only works when the audit receiver and the resolver are
 	// the same process, and that has to fail loudly rather than degrade into silent attribution
 	// loss.
-	replicaCount         int
-	branchBufferMaxBytes int64
-	sensitiveResources   types.SensitiveResourcePolicy
-	sshHostKeys          git.SSHHostKeyConfig
-	credentialPolicy     git.CredentialTransportPolicy
+	replicaCount       int
+	branchWorkerLimits git.BranchWorkerLimits
+	sensitiveResources types.SensitiveResourcePolicy
+	sshHostKeys        git.SSHHostKeyConfig
+	credentialPolicy   git.CredentialTransportPolicy
 	// sourceClusterQPS / sourceClusterBurst bound the rate at which the operator talks to a
 	// source cluster reached through a GitTarget.spec.kubeConfig. A remote is reached over a
 	// network the in-cluster config is not, so it carries client-side throttling by default.
@@ -614,7 +614,19 @@ func parseFlagsWithArgs(fs *flag.FlagSet, args []string) (appConfig, error) {
 	var branchBufferMaxSizeFlag string
 	fs.StringVar(&branchBufferMaxSizeFlag, "branch-buffer-max-size", branchBufferMaxSizeStr,
 		"Maximum in-memory event buffer per branch worker, as a Kubernetes resource quantity "+
-			"(e.g. 8Mi, 1Gi; default 8Mi). Bounds pod memory under bursty workloads; not user-facing.")
+			"(e.g. 8Mi, 1Gi; default 8Mi). Bounds the open commit window plus the writes retained "+
+			"for replay until a push succeeds. It does NOT bound the event queue, which is "+
+			"accounted only after a dequeue: see --branch-worker-queue-depth.")
+	fs.IntVar(&cfg.branchWorkerLimits.QueueDepth, "branch-worker-queue-depth",
+		git.DefaultBranchWorkerQueueDepth,
+		fmt.Sprintf("Event queue depth per branch worker (default %d). ", git.DefaultBranchWorkerQueueDepth)+
+			"A branch worker is shared by every "+
+			"GitTarget writing to one (GitProvider, branch), and a full queue DROPS the write "+
+			"(git_queue_drops_total) rather than stalling the watch path behind a slow remote, "+
+			"losing that live attributed commit even though convergence later heals the mirror. "+
+			"Size it so a bounded burst cannot overrun it: roughly (concurrent writers) x "+
+			"(GitTargets sharing the branch worker). Costs queue depth x payload of pod memory "+
+			"ON TOP of --branch-buffer-max-size, which does not cover this queue.")
 	var additionalSensitiveResources string
 	fs.StringVar(
 		&additionalSensitiveResources,
@@ -674,10 +686,18 @@ func parseFlagsWithArgs(fs *flag.FlagSet, args []string) (appConfig, error) {
 	if err != nil {
 		return appConfig{}, fmt.Errorf("invalid --branch-buffer-max-size %q: %w", branchBufferMaxSizeFlag, err)
 	}
-	cfg.branchBufferMaxBytes, _ = bufferQuantity.AsInt64()
-	if cfg.branchBufferMaxBytes <= 0 {
+	cfg.branchWorkerLimits.MaxBufferBytes, _ = bufferQuantity.AsInt64()
+	if cfg.branchWorkerLimits.MaxBufferBytes <= 0 {
 		return appConfig{}, fmt.Errorf("--branch-buffer-max-size must be > 0, got %s", branchBufferMaxSizeFlag)
 	}
+	// Refused rather than defaulted: 0 is a channel with no buffer, which drops every write
+	// that does not find the loop already waiting. An operator who typed it meant something
+	// else, and the symptom -- near-total silent loss under any load -- reads like a bug in
+	// the watch path rather than like the number they set.
+	if cfg.branchWorkerLimits.QueueDepth <= 0 {
+		return appConfig{}, fmt.Errorf(
+			"--branch-worker-queue-depth must be > 0, got %d", cfg.branchWorkerLimits.QueueDepth)
+	}
 
 	cfg.sensitiveResources, err = types.ParseSensitiveResourcePolicy(additionalSensitiveResources)
 	if err != nil {

cmd/main_branch_worker_limits_test.go

@@ -0,0 +1,65 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"flag"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ConfigButler/gitops-reverser/internal/git"
+)
+
+func parseLimitsArgs(t *testing.T, args ...string) (appConfig, error) {
+	t.Helper()
+	return parseFlagsWithArgs(flag.NewFlagSet("manager", flag.ContinueOnError), args)
+}
+
+// The queue depth was a compile-time constant, and the deployments that overrun it are the
+// ones whose shape the constant was never sized against: a branch worker is shared by every
+// GitTarget on a (GitProvider, branch), so the headroom an operator needs is a property of
+// their cluster, not of this binary. Assert the knob exists and carries a value through.
+func TestParseFlags_BranchWorkerQueueDepthIsConfigurable(t *testing.T) {
+	cfg, err := parseLimitsArgs(t, "--branch-worker-queue-depth=4096")
+	require.NoError(t, err)
+	require.Equal(t, 4096, cfg.branchWorkerLimits.QueueDepth)
+}
+
+func TestParseFlags_BranchWorkerLimitsDefault(t *testing.T) {
+	cfg, err := parseLimitsArgs(t)
+	require.NoError(t, err)
+	require.Equal(t, git.DefaultBranchWorkerQueueDepth, cfg.branchWorkerLimits.QueueDepth,
+		"an unset queue size must take the package default, not zero")
+	require.Equal(t, git.DefaultBranchBufferMaxBytes, cfg.branchWorkerLimits.MaxBufferBytes)
+}
+
+// Zero is refused rather than defaulted. An unbuffered channel drops every write that does
+// not find the event loop already waiting, which presents as near-total silent loss in the
+// watch path rather than as the number the operator typed.
+func TestParseFlags_BranchWorkerQueueDepthRejectsNonPositive(t *testing.T) {
+	for _, arg := range []string{"--branch-worker-queue-depth=0", "--branch-worker-queue-depth=-1"} {
+		_, err := parseLimitsArgs(t, arg)
+		require.ErrorContains(t, err, "--branch-worker-queue-depth must be > 0", "arg %q", arg)
+	}
+}
+
+// The depth is deliberately flag-only, with no environment fallback. Every install path
+// this repo ships configures the manager by argv -- the chart renders the flag
+// unconditionally, and the config/ manifests pass args too -- so an env seam here would be
+// a second configuration path that no install could actually reach, and one whose parse
+// order could fail a startup the explicit flag had already settled. Asserted so it does not
+// quietly come back.
+func TestParseFlags_BranchWorkerQueueDepthIgnoresEnvironment(t *testing.T) {
+	t.Setenv("BRANCH_WORKER_QUEUE_DEPTH", "2500")
+	cfg, err := parseLimitsArgs(t)
+	require.NoError(t, err)
+	require.Equal(t, git.DefaultBranchWorkerQueueDepth, cfg.branchWorkerLimits.QueueDepth,
+		"the depth is flag-only; an env var must not set it")
+
+	// An unparseable value cannot fail a startup, because nothing reads it.
+	t.Setenv("BRANCH_WORKER_QUEUE_DEPTH", "not-a-number")
+	cfg, err = parseLimitsArgs(t, "--branch-worker-queue-depth=77")
+	require.NoError(t, err)
+	require.Equal(t, 77, cfg.branchWorkerLimits.QueueDepth)
+}

cmd/main_chart_args_test.go

@@ -11,6 +11,8 @@ import (
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	"sigs.k8s.io/yaml"
+
+	"github.com/ConfigButler/gitops-reverser/internal/git"
 )
 
 // chartInputs are read by the test process itself so `go test` records them as cache
@@ -69,9 +71,13 @@ func TestChartRendersArgsTheBinaryAccepts(t *testing.T) {
 		wantAuditTLSCert string
 		wantHTTP2        bool
 		wantGitHTTP      bool
+		wantQueueDepth   int
+		wantBufferBytes  int64
 	}{
 		"chart defaults": {
-			wantKeyPrefix: "gitops-reverser",
+			wantKeyPrefix:   "gitops-reverser",
+			wantQueueDepth:  git.DefaultBranchWorkerQueueDepth,
+			wantBufferBytes: git.DefaultBranchBufferMaxBytes,
 		},
 		// servers.enableHTTP2 was a value no template read: setting it changed nothing, while
 		// its comment promised an HTTP/2 Rapid-Reset mitigation. Assert it reaches the binary.
@@ -124,6 +130,18 @@ func TestChartRendersArgsTheBinaryAccepts(t *testing.T) {
 			wantKeyPrefix: "gitops-reverser",
 			wantGitHTTP:   true,
 		},
+		// The branch-worker capacity knobs exist BECAUSE one compile-time value cannot be
+		// right for every deployment, so an operator raising them is the whole point of the
+		// flags: render an override and confirm it survives the trip to the binary.
+		"raised branch-worker capacity": {
+			setValues: []string{
+				"controllerManager.branchWorkerQueueDepth=4000",
+				"controllerManager.branchBufferMaxSize=32Mi",
+			},
+			wantKeyPrefix:   "gitops-reverser",
+			wantQueueDepth:  4000,
+			wantBufferBytes: 32 * 1024 * 1024,
+		},
 	}
 
 	for name, tc := range tests {
@@ -150,6 +168,12 @@ func TestChartRendersArgsTheBinaryAccepts(t *testing.T) {
 				"servers.enableHTTP2 must reach the binary, and default to off")
 			require.Equal(t, tc.wantGitHTTP, cfg.credentialPolicy.AllowInsecureGitHTTP,
 				"controllerManager.allowInsecureGitHTTP must reach the binary")
+			if tc.wantQueueDepth != 0 {
+				require.Equal(t, tc.wantQueueDepth, cfg.branchWorkerLimits.QueueDepth,
+					"controllerManager.branchWorkerQueueDepth must reach the binary")
+				require.Equal(t, tc.wantBufferBytes, cfg.branchWorkerLimits.MaxBufferBytes,
+					"controllerManager.branchBufferMaxSize must reach the binary")
+			}
 		})
 	}
 }

docs/interpreting-metrics.md

@@ -209,6 +209,44 @@ boundary, the commit, the push. Background:
 | `placement_kustomization_entries_total` | counter | `outcome`, `gittarget_namespace`, `gittarget_name` | `added` / `no_change` / `failed`. |
 | `git_resync_failures_total` | counter | `gittarget_namespace`, `gittarget_name` | Rule-change resyncs whose apply failed **after** enqueue. |
 
+### Sizing the branch worker queue against `git_queue_drops_total`
+
+`git_queue_drops_total{kind="write"}` is the one loss the end state hides. The enqueue onto a
+branch worker is deliberately non-blocking, so a full queue throws the write away rather than
+stalling the watch path behind a slow remote; the next resync's mark-and-sweep then heals the
+mirror, and the target goes on reporting `RenderMatchesLive=True`. What does not come back is the
+live, attributed commit for the dropped write — the change happened, and the history does not say
+so. Treat any sustained increment as a signal to raise
+`controllerManager.branchWorkerQueueDepth`.
+
+**The depth gauge will not warn you first.** `git_queue_depth` shows the traffic you have, and the
+bursts that overrun the queue are usually the traffic you do not: an unpaced `kubectl delete` of a
+few hundred objects arrives orders of magnitude faster than the application writes the same objects
+did. A workload that sits comfortably at a fraction of the depth can still drop under an
+administrative one.
+
+**Size it by burst, not by rate.** A queue slot holds one write request, so what has to fit is the
+number of concurrent write requests a bounded burst can produce — roughly
+`(concurrent writers) × (GitTargets sharing the branch worker)`. A burst that cannot exceed the
+depth cannot drop at all, whatever its arrival shape. Note the second factor: workers are keyed by
+`(GitProvider namespace, GitProvider name, branch)`, so pointing a second GitTarget at a branch
+halves the headroom of a worker already in use, and nothing in the API surface says so. Confirm
+which targets share one worker by joining on `git_branch_targets` — several targets against a
+single `git_queue_depth` series is one worker serving all of them.
+
+**What it costs.** `--branch-buffer-max-size` does **not** cover this queue. That cap bounds the
+open commit window plus the writes retained for replay until a push succeeds, and it is accounted
+only once an item *leaves* the queue — so queue depth × payload is additional pod memory, on top of
+it.
+
+Budget it as **queue depth × serialized size × ~6**. A queued object is held as an unstructured
+map, which measures around six times the bytes it serializes to; the channel itself is negligible
+beside that. So a full queue of 1000 at the ~800-byte resources of a small CRD is about 5 MiB per
+saturated worker, against a 1Gi default limit — but the multiplier is on the payload, so a folder
+mirroring megabyte-sized ConfigMaps or Secrets reaches gigabytes at the same depth. Size against
+your own payloads, and against how many workers could saturate at once: the figure is per worker,
+and workers are per `(GitProvider, branch)`.
+
 ### Outcomes are classified, not ranked
 
 Several values above are the pipeline working, and a dashboard that paints every non-happy outcome

internal/controller/suite_test.go

@@ -99,7 +99,7 @@ var _ = BeforeSuite(func() {
 
 	// Initialize WorkerManager for new architecture
 	workerManager := git.NewWorkerManager(
-		mgr.GetClient(), logf.Log.WithName("worker-manager"), 0, types.SensitiveResourcePolicy{},
+		mgr.GetClient(), logf.Log.WithName("worker-manager"), git.BranchWorkerLimits{}, types.SensitiveResourcePolicy{},
 	)
 	err = mgr.Add(workerManager)
 	Expect(err).NotTo(HaveOccurred())

internal/git/branch_worker.go

@@ -32,9 +32,6 @@ import (
 )
 
 const (
-	// branchWorkerQueueSize is the size of the event queue for each branch worker.
-	branchWorkerQueueSize = 100
-
 	// metadataCacheDuration is how long metadata is considered fresh before re-fetching.
 	// This optimization prevents redundant Git fetches when multiple GitTargets
 	// share the same branch and reconcile within a short time window.
@@ -226,23 +223,20 @@ const (
 	windowFinalizeReasonShutdown          windowFinalizeReason = "shutdown"
 )
 
-// NewBranchWorker creates a worker for a (provider, branch) combination.
-// Pass 0 (or a negative value) for branchBufferMaxBytes to use
-// DefaultBranchBufferMaxBytes.
+// NewBranchWorker creates a worker for a (provider, branch) combination. A zero
+// BranchWorkerLimits gets the defaults for both knobs; see BranchWorkerLimits.
 func NewBranchWorker(
 	client client.Client,
 	log logr.Logger,
 	providerName, providerNamespace string,
 	branch string,
 	writer *contentWriter,
-	branchBufferMaxBytes int64,
+	limits BranchWorkerLimits,
 ) *BranchWorker {
 	if writer == nil {
 		writer = newContentWriter(itypes.SensitiveResourcePolicy{})
 	}
-	if branchBufferMaxBytes <= 0 {
-		branchBufferMaxBytes = DefaultBranchBufferMaxBytes
-	}
+	limits = limits.withDefaults()
 	return &BranchWorker{
 		GitProviderRef:       providerName,
 		GitProviderNamespace: providerNamespace,
@@ -254,9 +248,9 @@ func NewBranchWorker(
 			"branch", branch,
 		),
 		contentWriter:        writer,
-		eventQueue:           make(chan WorkItem, branchWorkerQueueSize),
+		eventQueue:           make(chan WorkItem, limits.QueueDepth),
 		pendingResyncs:       make(map[resyncKey]*pendingResync),
-		branchBufferMaxBytes: branchBufferMaxBytes,
+		branchBufferMaxBytes: limits.MaxBufferBytes,
 	}
 }
 
@@ -734,7 +728,7 @@ func (w *BranchWorker) processEvents() {
 	loop := newBranchWorkerEventLoop(w, DefaultCommitWindow)
 	w.Log.Info("Branch worker event loop configured",
 		"defaultCommitWindow", DefaultCommitWindow.String(),
-		"queueSize", cap(w.eventQueue),
+		"queueDepth", cap(w.eventQueue),
 		"branchBufferMaxBytes", w.branchBufferMaxBytes)
 	loop.run()
 }

internal/git/branch_worker_credread_test.go

@@ -73,7 +73,7 @@ func TestCommitPendingWrites_ResolvesCredentialsOncePerPushCycle(t *testing.T) {
 		}).
 		Build()
 
-	worker := NewBranchWorker(countingClient, logr.Discard(), "test-repo", "default", "main", nil, 0)
+	worker := NewBranchWorker(countingClient, logr.Discard(), "test-repo", "default", "main", nil, BranchWorkerLimits{})
 	worker.ctx = context.Background()
 
 	// First commit of the cycle: fetches the remote tip, so it resolves credentials exactly once.