🐛 fix(sink): keep one inventory's ownership conflicts out of the shared breaker

docs/adr/0602-error-taxonomy.md

@@ -36,6 +36,12 @@ Additional rules:
 - No `panic` in reconcilers; entrypoint `recover` → requeue + Event.
 - Context deadlines on all external calls; never log secrets or full payloads.
 - Circuit breaker (`gobreaker`) per sink/doc backend for `ErrTransient` storms.
+  *Note 2026-11-05:* the breaker counts more than `ErrTransient`. Every failure the sink
+  backend returns while writing an export counts (transient, forbidden, unclassified, and
+  sink-wide terminal errors such as Git auth or protected-branch rejections), except a result
+  marked inventory-scoped (today a Git ownership conflict on one inventory's own paths), which
+  counts neither as a failure nor as a success. The decision above is unchanged; see the
+  troubleshooting guide for thresholds and recovery.
 
 Conditions follow Kubernetes conventions: `Ready`, `Synced`, `Degraded` with `observedGeneration`.
 

docs/operator-manual/troubleshooting.md

@@ -211,7 +211,7 @@ Family CRDs: **`KollectSnapshotSink`** (git/gitlab/s3/gcs), **`KollectDatabaseSi
 | Postgres connection failures | DSN, network policy, TLS, pool timeout | Database sink conditions; `transient` sink errors; connection test metric | Verify Secret keys, egress NetworkPolicy, server reachable | DBA for server-side limits |
 | Postgres rows stale while status looks healthy | Upsert-only drift, or an export error on that ref | Compare `lastExportTime` on the database entry in `status.sinkExports[]` | Follow [Postgres state store — Troubleshooting](../examples/postgres-state-store.md#troubleshooting) | DBA for server-side limits |
 | S3/GCS 403 | IAM, wrong bucket, signature | Export logs; `terminal`/`forbidden` | Fix credentials and bucket policy | Cloud IAM review |
-| `sink circuit breaker open` in logs | 5 consecutive transient failures per sink key | `transient` errors then silence ~30s | Fix backend; breaker self-closes after timeout | Backend SLA breach |
+| `sink circuit breaker open` in logs | 5 consecutive counted backend write failures to one sink (breaker key `<sink namespace>/<sink name>`) within one 60 s counting interval (the counts reset every 60 s while the breaker is closed). Counted: transient, forbidden and unclassified errors, and sink-wide terminal errors such as a Git auth or protected-branch rejection. A failure about one inventory (an ownership conflict on its own paths) does not count; it still fails that inventory's export (status, Warning event, logs) | Sink errors, then `sink circuit breaker open for "<namespace>/<name>"` (transient) for ~30 s; exports to that sink do not reach the backend while it is open | Fix the backend. After 30 s one trial export is admitted: success closes the breaker, a counted failure re-opens it for another 30 s, an inventory-scoped result (ownership conflict) leaves it half-open so the next export is the trial | Backend SLA breach |
 | `SinkReachable=False`, `SinkNotFound` | Wrong sink name or cross-namespace ref in `*SinkRefs` | Inventory message; `kubectl get kollect*sink -n <inv-ns>` | Fix ref name; create sink in inventory namespace | — |
 | `SinkReachable=False`, `SinkUnreachable` | Backend down despite CR present; bad DSN or TLS failure | Sink `ConnectionVerified`; probe annotation | Fix network/credentials first | — |
 | Inventory / namespace stuck `Terminating` | Sink cleanup failed terminally during deletion; finalizer retained | Events reason `CleanupTerminal` on the inventory; `increase(kollect_cleanup_terminal_total[1h]) > 0`; Degraded reason `CleanupTerminal` | Fix the sink backend and let the next reconcile finish cleanup — the terminal wedge re-checks every 5 minutes, so a Secret-only credential fix (no watch fires on Secret edits) clears within minutes; or accept the exported objects and set `kollect.dev/force-cleanup: "true"` on the deleting object to drop the finalizer | — |
@@ -325,7 +325,7 @@ Structured controller logs (`logr`). Grep operator pod logs (namespace typically
 | `export failed` | Sink export path |
 | `export payload exceeds spill warn threshold` | Approaching 1 MiB — shard soon |
 | `debounced` | Export skipped by interval/coalesce |
-| `sink circuit breaker open` | Repeated transient sink failures |
+| `sink circuit breaker open` | Repeated sink failures (transient, forbidden or sink-wide terminal); see *Export — sink backends* |
 | `reconcile panic recovered` | Panic converted to requeue (EC-P2-01) |
 | `git push` / `git auth failed` | Snapshot sink transport |
 

internal/controller/cluster_inventory_finalizer_test.go

@@ -27,6 +27,7 @@ import (
 
 func TestKollectClusterInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	scheme := runtime.NewScheme()
 	if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -37,7 +38,7 @@ func TestKollectClusterInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "kollect-system"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "kollect-system"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.DatabaseSinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -55,7 +56,7 @@ func TestKollectClusterInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer
 			DeletionTimestamp: &now,
 		},
 		Spec: kollectdevv1alpha1.KollectClusterInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 
@@ -113,6 +114,7 @@ func TestKollectClusterInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer
 // 12-hour informer resync.
 func TestKollectClusterInventoryReconciler_terminalCleanupDoesNotRequeue(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	scheme := runtime.NewScheme()
 	if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -123,7 +125,7 @@ func TestKollectClusterInventoryReconciler_terminalCleanupDoesNotRequeue(t *test
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "kollect-system"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "kollect-system"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.DatabaseSinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -141,7 +143,7 @@ func TestKollectClusterInventoryReconciler_terminalCleanupDoesNotRequeue(t *test
 			DeletionTimestamp: &now,
 		},
 		Spec: kollectdevv1alpha1.KollectClusterInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 
@@ -277,6 +279,7 @@ func TestKollectClusterInventoryReconciler_deleteWithMissingSinkRemovesFinalizer
 // EC-P1-03 counterpart: transient cleanup failures keep the error-driven retry.
 func TestKollectClusterInventoryReconciler_transientCleanupStillRetries(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	scheme := runtime.NewScheme()
 	if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -287,7 +290,7 @@ func TestKollectClusterInventoryReconciler_transientCleanupStillRetries(t *testi
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "kollect-system"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "kollect-system"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.DatabaseSinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -305,7 +308,7 @@ func TestKollectClusterInventoryReconciler_transientCleanupStillRetries(t *testi
 			DeletionTimestamp: &now,
 		},
 		Spec: kollectdevv1alpha1.KollectClusterInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 

internal/controller/kollectclusterinventory_debounce_checksum_test.go

@@ -208,7 +208,10 @@ func TestClusterInventory_CeilingChangeWithinDebounceWindowReExports(t *testing.
 // export must increment kollect_export_debounced_total for KollectClusterInventory, so cluster
 // debounce decisions are visible to dashboards/alerts exactly like the namespaced path.
 func TestClusterInventory_DebouncedExportIncrementsMetric(t *testing.T) {
-	t.Parallel()
+	// Not parallel: the assertion reads the process-global
+	// kollect_export_debounced_total counter as a delta, so no other test may
+	// increment the same {controller=KollectClusterInventory} label
+	// concurrently (a parallel debounced cluster export made the delta 2).
 
 	const (
 		sinkNS = sink.DefaultSecretNamespace

internal/controller/kollectinventory_export_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
@@ -238,6 +239,7 @@ func TestKollectInventoryReconciler_recordsLastExportPaths(t *testing.T) {
 
 func TestKollectInventoryReconciler_exportsDeploymentToSink(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	store := collect.NewStore()
 	store.Upsert(collect.Item{
@@ -260,7 +262,7 @@ func TestKollectInventoryReconciler_exportsDeploymentToSink(t *testing.T) {
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.DatabaseSinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -273,7 +275,7 @@ func TestKollectInventoryReconciler_exportsDeploymentToSink(t *testing.T) {
 	inv := &kollectdevv1alpha1.KollectInventory{
 		ObjectMeta: metav1.ObjectMeta{Name: "team-inventory", Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 
@@ -325,7 +327,7 @@ func TestKollectInventoryReconciler_exportsDeploymentToSink(t *testing.T) {
 	if len(got.Status.SinkExports) != 1 {
 		t.Fatalf("SinkExports = %d, want 1", len(got.Status.SinkExports))
 	}
-	wantName := string(kollectdevv1alpha1.SinkFamilyDatabase) + "/postgres-demo"
+	wantName := string(kollectdevv1alpha1.SinkFamilyDatabase) + "/" + sinkName
 	if got.Status.SinkExports[0].Name != wantName {
 		t.Fatalf("sink export name = %q, want %s", got.Status.SinkExports[0].Name, wantName)
 	}
@@ -334,10 +336,25 @@ func TestKollectInventoryReconciler_exportsDeploymentToSink(t *testing.T) {
 	}
 }
 
+// sinkNameSeq makes sink names unique per test invocation. The sink circuit
+// breaker is process-wide and keyed by sink namespace/name, so tests that
+// share a sink name share one breaker: parallel siblings, and -count=N reruns,
+// add their failures to it and can open it for a test that expects an export.
+var sinkNameSeq atomic.Uint64
+
+// uniqueSinkName returns base with a per-invocation suffix, for a sink whose
+// breaker no other test (or earlier run of this test) touches.
+func uniqueSinkName(t *testing.T, base string) string {
+	t.Helper()
+
+	return fmt.Sprintf("%s-%d", base, sinkNameSeq.Add(1))
+}
+
 // newExportFailureReconciler builds a reconciler whose single postgres sink
 // always fails export with exportErr.
 func newExportFailureReconciler(t *testing.T, exportErr error) (*KollectInventoryReconciler, client.Client) {
 	t.Helper()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	store := collect.NewStore()
 	store.Upsert(collect.Item{
@@ -359,7 +376,7 @@ func newExportFailureReconciler(t *testing.T, exportErr error) (*KollectInventor
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.DatabaseSinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -372,7 +389,7 @@ func newExportFailureReconciler(t *testing.T, exportErr error) (*KollectInventor
 	inv := &kollectdevv1alpha1.KollectInventory{
 		ObjectMeta: metav1.ObjectMeta{Name: "team-inventory", Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 

internal/controller/kollectinventory_finalizer_test.go

@@ -103,6 +103,7 @@ func TestKollectInventoryReconciler_addsCleanupFinalizer(t *testing.T) {
 
 func TestKollectInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	store := collect.NewStore()
 	store.Upsert(collect.Item{
@@ -124,7 +125,7 @@ func TestKollectInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer(t *tes
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.SinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -143,7 +144,7 @@ func TestKollectInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer(t *tes
 			DeletionTimestamp: &now,
 		},
 		Spec: kollectdevv1alpha1.KollectInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 
@@ -200,6 +201,7 @@ func TestKollectInventoryReconciler_deleteExportsEmptyAndRemovesFinalizer(t *tes
 // result) and must keep the finalizer in place for manual intervention.
 func TestKollectInventoryReconciler_terminalCleanupDoesNotRequeue(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	scheme := runtime.NewScheme()
 	if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -210,7 +212,7 @@ func TestKollectInventoryReconciler_terminalCleanupDoesNotRequeue(t *testing.T)
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.SinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -229,7 +231,7 @@ func TestKollectInventoryReconciler_terminalCleanupDoesNotRequeue(t *testing.T)
 			DeletionTimestamp: &now,
 		},
 		Spec: kollectdevv1alpha1.KollectInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 
@@ -371,6 +373,7 @@ func TestKollectInventoryReconciler_deleteWithMissingSinkRemovesFinalizer(t *tes
 // EC-P1-03 counterpart: transient cleanup failures keep the error-driven retry.
 func TestKollectInventoryReconciler_transientCleanupStillRetries(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	scheme := runtime.NewScheme()
 	if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -381,7 +384,7 @@ func TestKollectInventoryReconciler_transientCleanupStillRetries(t *testing.T) {
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.SinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -400,7 +403,7 @@ func TestKollectInventoryReconciler_transientCleanupStillRetries(t *testing.T) {
 			DeletionTimestamp: &now,
 		},
 		Spec: kollectdevv1alpha1.KollectInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 
@@ -1053,6 +1056,7 @@ func TestKollectInventoryReconciler_terminalCleanupIncrementsMetric(t *testing.T
 // K-30 escape hatch: force-cleanup drops the finalizer without backend contact.
 func TestKollectInventoryReconciler_forceCleanupAnnotationDropsFinalizer(t *testing.T) {
 	t.Parallel()
+	sinkName := uniqueSinkName(t, "postgres-demo")
 
 	scheme := runtime.NewScheme()
 	if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -1063,7 +1067,7 @@ func TestKollectInventoryReconciler_forceCleanupAnnotationDropsFinalizer(t *test
 	}
 
 	sinkObj := &kollectdevv1alpha1.KollectDatabaseSink{
-		ObjectMeta: metav1.ObjectMeta{Name: "postgres-demo", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: sinkName, Namespace: "default"},
 		Spec: kollectdevv1alpha1.KollectDatabaseSinkSpec{
 			Type: kollectdevv1alpha1.SinkTypePostgres,
 			Postgres: &kollectdevv1alpha1.PostgresSpec{
@@ -1084,7 +1088,7 @@ func TestKollectInventoryReconciler_forceCleanupAnnotationDropsFinalizer(t *test
 			},
 		},
 		Spec: kollectdevv1alpha1.KollectInventorySpec{
-			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList("postgres-demo"),
+			DatabaseSinkRefs: kollectdevv1alpha1.NewSinkRefList(sinkName),
 		},
 	}
 	pgSecret := &corev1.Secret{

internal/errors/errors.go

@@ -147,3 +147,34 @@ func Format(class error, format string, args ...any) error {
 
 	return &ClassError{Class: class, Err: msg}
 }
+
+// ErrInventoryScoped marks an export failure that is about one inventory (its own paths or its
+// ownership claim), not about the sink. The per-sink circuit breaker does not count it.
+var ErrInventoryScoped = errors.New("inventory-scoped")
+
+// InventoryScoped marks err as inventory-scoped and always makes it Terminal, whatever class err
+// carried before: ClassOf and IsTerminal report Terminal for a Transient or Forbidden input too (the
+// outer class wins; the inner class sentinel stays reachable through errors.Is). The result keeps
+// err's text and its errors.Is/errors.As chain, so status messages and logs do not change.
+func InventoryScoped(err error) error {
+	if err == nil {
+		return nil
+	}
+
+	return Terminal(&inventoryScopedError{err: err})
+}
+
+// IsInventoryScoped reports whether err carries the inventory-scoped marker.
+func IsInventoryScoped(err error) bool {
+	return errors.Is(err, ErrInventoryScoped)
+}
+
+type inventoryScopedError struct {
+	err error
+}
+
+func (e *inventoryScopedError) Error() string { return e.err.Error() }
+
+func (e *inventoryScopedError) Unwrap() error { return e.err }
+
+func (e *inventoryScopedError) Is(target error) bool { return target == ErrInventoryScoped }