From 08e8889c207fbf31fc20450670ac7bb18b18cf11 Mon Sep 17 00:00:00 2001 From: Konrad Heimel Date: Sat, 10 Oct 2026 17:30:56 +0200 Subject: [PATCH 1/3] :white_check_mark: test(controller): wait for the delete-watch handler before deleting The envtest spec deleted its sink as soon as the cache synced, but the manager attaches a controller's Watches handler asynchronously after that. A delete landing before registration is never replayed to the handler (late handlers only get synthetic Adds), so the spy saw zero closes and the spec timed out. Probe with throwaway sinks until one delete reaches the seam, then run the real assertion. --- .../family_sink_delete_watch_envtest_test.go | 72 +++++++++++++------ 1 file changed, 50 insertions(+), 22 deletions(-) diff --git a/internal/controller/family_sink_delete_watch_envtest_test.go b/internal/controller/family_sink_delete_watch_envtest_test.go index 3274dd45..426b76a4 100644 --- a/internal/controller/family_sink_delete_watch_envtest_test.go +++ b/internal/controller/family_sink_delete_watch_envtest_test.go @@ -52,31 +52,59 @@ var _ = Describe("family-sink delete watch (envtest, BEP-1)", func() { } Expect(k8sClient.Create(watchCtx, sinkObj)).To(Succeed()) - // The watch informer must know the sink before it is deleted, so the - // delete lands as a normal delete event with the object's UID. - Eventually(func() error { - return mgr.GetClient().Get(watchCtx, types.NamespacedName{Namespace: sinkObj.Namespace, Name: sinkObj.Name}, &kollectdevv1alpha1.KollectSnapshotSink{}) - }).WithTimeout(30 * time.Second).Should(Succeed()) - - spy := &deleteWatchSpyBackend{} - reg := sink.NewRegistry() - reg.Register("s3", func(_ kollectdevv1alpha1.KollectSinkSpec, _ sink.BuildContext) (sink.Backend, error) { - return spy, nil - }) envelope, envErr := export.MarshalEnvelope([]collect.Item{}, export.Metadata{}) Expect(envErr).NotTo(HaveOccurred()) - _, poolErr := sink.RunExportEnvelope(sink.ExportEnvelopeRequest{ - Ctx: watchCtx, - Client: k8sClient, - Registry: reg, - SinkNamespace: sinkObj.Namespace, - SinkName: sinkObj.Name, - SinkUID: sinkObj.UID, - Envelope: envelope, - SinkSpec: sinkObj.FamilySinkSpec(), - }) - Expect(poolErr).NotTo(HaveOccurred()) + // createPooledSink creates a sink, waits until the manager's cache has + // it and pools a spy backend for its UID. + createPooledSink := func() (*kollectdevv1alpha1.KollectSnapshotSink, *deleteWatchSpyBackend) { + obj := &kollectdevv1alpha1.KollectSnapshotSink{ + ObjectMeta: metav1.ObjectMeta{GenerateName: "delete-watch-", Namespace: "default"}, + Spec: kollectdevv1alpha1.KollectSnapshotSinkSpec{Type: "s3"}, + } + Expect(k8sClient.Create(watchCtx, obj)).To(Succeed()) + Eventually(func() error { + return mgr.GetClient().Get(watchCtx, types.NamespacedName{Namespace: obj.Namespace, Name: obj.Name}, &kollectdevv1alpha1.KollectSnapshotSink{}) + }).WithTimeout(30 * time.Second).Should(Succeed()) + + spy := &deleteWatchSpyBackend{} + reg := sink.NewRegistry() + reg.Register("s3", func(_ kollectdevv1alpha1.KollectSinkSpec, _ sink.BuildContext) (sink.Backend, error) { + return spy, nil + }) + _, poolErr := sink.RunExportEnvelope(sink.ExportEnvelopeRequest{ + Ctx: watchCtx, + Client: k8sClient, + Registry: reg, + SinkNamespace: obj.Namespace, + SinkName: obj.Name, + SinkUID: obj.UID, + Envelope: envelope, + SinkSpec: obj.FamilySinkSpec(), + }) + Expect(poolErr).NotTo(HaveOccurred()) + + return obj, spy + } + + // Cache sync does not mean the controller's Watches handler is attached: + // the manager starts controllers (and their event sources) after the + // caches sync, asynchronously. A delete issued before the handler is + // registered is never replayed to it (late handlers only get synthetic + // Adds for objects still in the store), so the spec would flake with + // closes=0. Probe with throwaway sinks until one delete reaches the + // seam: from then on the handler is provably live. + watchLive := false + for deadline := time.Now().Add(60 * time.Second); !watchLive && time.Now().Before(deadline); { + probe, probeSpy := createPooledSink() + Expect(k8sClient.Delete(watchCtx, probe)).To(Succeed()) + for end := time.Now().Add(2 * time.Second); !watchLive && time.Now().Before(end); time.Sleep(50 * time.Millisecond) { + watchLive = probeSpy.closes.Load() == 1 + } + } + Expect(watchLive).To(BeTrue(), "delete watch handler never became live") + + sinkObj, spy := createPooledSink() Expect(k8sClient.Delete(watchCtx, sinkObj)).To(Succeed()) From 44d8a4a61d05b6f7207b40f7d9a6474234a184e4 Mon Sep 17 00:00:00 2001 From: Konrad Heimel Date: Sat, 10 Oct 2026 17:46:13 +0200 Subject: [PATCH 2/3] :bug: fix(git): match HTTP status codes in errors as whole tokens ClassifyExportError matched 401/503/429 as bare substrings, so digits in a random temp-dir path (a missing local remote) read as an auth failure and turned a transient clone error terminal. TestRunCleanupExport_ gitReleaseFailureIsRetried failed in CI whenever the dir name contained 401. Require a non-digit boundary around the code. --- internal/sink/git/errors.go | 30 ++++++++++++++++++++++++++---- internal/sink/git/errors_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/internal/sink/git/errors.go b/internal/sink/git/errors.go index 0e8d9707..19c087db 100644 --- a/internal/sink/git/errors.go +++ b/internal/sink/git/errors.go @@ -64,18 +64,40 @@ func isTransientTransportError(err error) bool { strings.Contains(msg, "timeout") || strings.Contains(msg, "temporary failure") || strings.Contains(msg, "eof") || - strings.Contains(msg, "503") || - strings.Contains(msg, "429") || + hasStatusCode(msg, "503") || + hasStatusCode(msg, "429") || strings.Contains(msg, "too many requests") } +// hasStatusCode reports whether msg contains code as a standalone token. A bare +// substring match misreads digits inside paths (a temp dir named ...3401029481 +// is not an HTTP 401), which flipped a missing-remote clone from transient to +// terminal. +func hasStatusCode(msg, code string) bool { + for from := 0; ; { + i := strings.Index(msg[from:], code) + if i < 0 { + return false + } + + start, end := from+i, from+i+len(code) + if (start == 0 || !isDigit(msg[start-1])) && (end == len(msg) || !isDigit(msg[end])) { + return true + } + + from = start + 1 + } +} + +func isDigit(b byte) bool { return b >= '0' && b <= '9' } + func isAuthFailure(msg string, err error) bool { if strings.Contains(msg, "authentication required") || strings.Contains(msg, "invalid credentials") || strings.Contains(msg, "authorization failed") || strings.Contains(msg, "access denied") || - strings.Contains(msg, "401") || - strings.Contains(msg, "403 forbidden") { + hasStatusCode(msg, "401") || + hasStatusCode(msg, "403 forbidden") { return true } diff --git a/internal/sink/git/errors_test.go b/internal/sink/git/errors_test.go index 1d6ca528..cc9ffd09 100644 --- a/internal/sink/git/errors_test.go +++ b/internal/sink/git/errors_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "strings" "syscall" "testing" @@ -151,3 +152,28 @@ func TestIsNonFastForwardError(t *testing.T) { t.Fatal("expected non-fast-forward for combined rejected-refs message") } } + +// Digits inside a path (a random temp dir) are not an HTTP status: a missing +// local remote must not classify as an auth failure. +func TestClassifyExportError_statusCodeNeedsTokenBoundary(t *testing.T) { + t.Parallel() + + pathErr := errors.New("git clone: fatal: '/tmp/x3401029481/missing.git' does not appear to be a git repository") + if kollecterrors.IsTerminal(ClassifyExportError(pathErr)) { + t.Fatalf("digits in a path classified as terminal: %v", ClassifyExportError(pathErr)) + } + + for _, msg := range []string{"unexpected client error: 401", "HTTP 403 Forbidden"} { + if !kollecterrors.IsTerminal(ClassifyExportError(errors.New(msg))) { + t.Fatalf("%q must stay terminal", msg) + } + } + + if !kollecterrors.IsTransient(ClassifyExportError(errors.New("server said 503"))) { + t.Fatal("standalone 503 must stay transient") + } + + if strings.Contains(ClassifyExportError(errors.New("path /tmp/a15039b")).Error(), "git transport") { + t.Fatal("503 inside a longer number must not be transient") + } +} From c5ad28f362e43e9b405fa1de63a88b169ead3a9d Mon Sep 17 00:00:00 2001 From: Konrad Heimel Date: Sat, 10 Oct 2026 18:33:50 +0200 Subject: [PATCH 3/3] :bug: fix(git): classify HTTP status codes from typed errors and status phrases Bare-substring and digit-boundary matching still misread codes inside URLs, hostnames and hex mirror dirs (team401, git-503.example.com, 3fa401bc9e0d). Read the status from go-git's typed http.Err and from status phrases only (status code: N, returned error: N, HTTP N, N forbidden, (N)), classify 401/403 terminal and 429/5xx transient, and treat ErrAuthorizationFailed as auth. The git CLI's returned error: 403 is now terminal too. The delete-watch envtest now gates on an observed reconcile of the sink (workers start after the Watches handler is registered) instead of probe deletes, and no longer leaks an unused sink. --- .../family_sink_delete_watch_envtest_test.go | 39 +++++----- internal/sink/git/errors.go | 77 ++++++++++++++----- internal/sink/git/errors_test.go | 64 ++++++++++----- 3 files changed, 121 insertions(+), 59 deletions(-) diff --git a/internal/controller/family_sink_delete_watch_envtest_test.go b/internal/controller/family_sink_delete_watch_envtest_test.go index 426b76a4..2a6525ff 100644 --- a/internal/controller/family_sink_delete_watch_envtest_test.go +++ b/internal/controller/family_sink_delete_watch_envtest_test.go @@ -9,6 +9,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "github.com/prometheus/client_golang/prometheus/testutil" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" @@ -16,6 +17,7 @@ import ( kollectdevv1alpha1 "github.com/platformrelay/kollect/api/v1alpha1" "github.com/platformrelay/kollect/internal/collect" "github.com/platformrelay/kollect/internal/export" + "github.com/platformrelay/kollect/internal/metrics" "github.com/platformrelay/kollect/internal/sink" ) @@ -46,12 +48,6 @@ var _ = Describe("family-sink delete watch (envtest, BEP-1)", func() { Eventually(func() bool { return mgr.GetCache().WaitForCacheSync(watchCtx) }). WithTimeout(30 * time.Second).WithPolling(200 * time.Millisecond).Should(BeTrue()) - sinkObj := &kollectdevv1alpha1.KollectSnapshotSink{ - ObjectMeta: metav1.ObjectMeta{GenerateName: "delete-watch-", Namespace: "default"}, - Spec: kollectdevv1alpha1.KollectSnapshotSinkSpec{Type: "s3"}, - } - Expect(k8sClient.Create(watchCtx, sinkObj)).To(Succeed()) - envelope, envErr := export.MarshalEnvelope([]collect.Item{}, export.Metadata{}) Expect(envErr).NotTo(HaveOccurred()) @@ -87,25 +83,26 @@ var _ = Describe("family-sink delete watch (envtest, BEP-1)", func() { return obj, spy } - // Cache sync does not mean the controller's Watches handler is attached: - // the manager starts controllers (and their event sources) after the - // caches sync, asynchronously. A delete issued before the handler is - // registered is never replayed to it (late handlers only get synthetic - // Adds for objects still in the store), so the spec would flake with - // closes=0. Probe with throwaway sinks until one delete reaches the - // seam: from then on the handler is provably live. - watchLive := false - for deadline := time.Now().Add(60 * time.Second); !watchLive && time.Now().Before(deadline); { - probe, probeSpy := createPooledSink() - Expect(k8sClient.Delete(watchCtx, probe)).To(Succeed()) - for end := time.Now().Add(2 * time.Second); !watchLive && time.Now().Before(end); time.Sleep(50 * time.Millisecond) { - watchLive = probeSpy.closes.Load() == 1 - } + const controllerLabel = "kollectsnapshotsink-delete-watch" + + reconciles := func() float64 { + return testutil.ToFloat64(metrics.ReconcileTotal.WithLabelValues(controllerLabel, metrics.ResultSuccess)) + + testutil.ToFloat64(metrics.ReconcileTotal.WithLabelValues(controllerLabel, metrics.ResultFailure)) } - Expect(watchLive).To(BeTrue(), "delete watch handler never became live") + before := reconciles() sinkObj, spy := createPooledSink() + // Cache sync does not mean the controller's Watches handler is attached: + // the manager starts controllers after the caches sync, asynchronously, + // and a delete issued before the handler registers is never replayed to + // it (late handlers only get synthetic Adds for objects still in the + // store). Workers start only after every source has synced, which is + // after its handler was added, so an observed reconcile of the sink + // proves the Watches handler is live before the delete. + Eventually(reconciles).WithTimeout(30*time.Second).WithPolling(50*time.Millisecond). + Should(BeNumerically(">", before), "controller never reconciled the sink: manager/controller did not start") + Expect(k8sClient.Delete(watchCtx, sinkObj)).To(Succeed()) // The delete event must reach the seam and Close the pooled backend — diff --git a/internal/sink/git/errors.go b/internal/sink/git/errors.go index 19c087db..3ba2755b 100644 --- a/internal/sink/git/errors.go +++ b/internal/sink/git/errors.go @@ -8,9 +8,13 @@ package git import ( "errors" "fmt" + "regexp" + "strconv" "strings" + "github.com/go-git/go-git/v5/plumbing" gittransport "github.com/go-git/go-git/v5/plumbing/transport" + githttp "github.com/go-git/go-git/v5/plumbing/transport/http" utilnet "k8s.io/apimachinery/pkg/util/net" kollecterrors "github.com/platformrelay/kollect/internal/errors" @@ -34,7 +38,7 @@ func ClassifyExportError(err error) error { if strings.Contains(msg, "protected branch") || strings.Contains(msg, "pre-receive hook declined") || - strings.Contains(msg, "remote rejected") && strings.Contains(msg, "403") { + strings.Contains(msg, "remote rejected") && hasAnyStatus(httpStatusCodes(err, msg), isAuthStatus) { return kollecterrors.Terminal(fmt.Errorf("git push rejected: %w", err)) } @@ -64,44 +68,77 @@ func isTransientTransportError(err error) bool { strings.Contains(msg, "timeout") || strings.Contains(msg, "temporary failure") || strings.Contains(msg, "eof") || - hasStatusCode(msg, "503") || - hasStatusCode(msg, "429") || + hasAnyStatus(httpStatusCodes(err, msg), isRetryableStatus) || strings.Contains(msg, "too many requests") } -// hasStatusCode reports whether msg contains code as a standalone token. A bare -// substring match misreads digits inside paths (a temp dir named ...3401029481 -// is not an HTTP 401), which flipped a missing-remote clone from transient to -// terminal. -func hasStatusCode(msg, code string) bool { - for from := 0; ; { - i := strings.Index(msg[from:], code) - if i < 0 { - return false +var ( + // statusPhraseRE matches a status code only after a phrase that introduces + // one ("status code: 502", "returned error: 403", "HTTP 401", "code=401"). + statusPhraseRE = regexp.MustCompile(`(?:returned error:|status code:?|status:|code[=:]|http/?[0-9.]*|error:)\s*\(?([0-9]{3})(?:[^0-9]|$)`) + // statusPunctRE matches a parenthesised or colon-terminated code: "(429)", "401:". + statusPunctRE = regexp.MustCompile(`(?:^|[\s(])([0-9]{3})[):]`) + // statusTextRE matches a code followed by its reason phrase: "403 forbidden". + statusTextRE = regexp.MustCompile(`(?:^|\s)([0-9]{3}) (?:unauthorized|forbidden|too many requests|service unavailable|bad gateway|gateway timeout|internal server error)`) +) + +// httpStatusCodes extracts the HTTP status codes an export error carries. It +// prefers go-git's typed *http.Err (its Response.StatusCode is authoritative) +// and otherwise reads only status phrases, never bare digit runs: a code-like +// substring in a URL, hostname or temp/mirror directory name (team401, +// git-503.example.com, a hex mirror dir) is not a status. +func httpStatusCodes(err error, msg string) []int { + var codes []int + + var httpErr *githttp.Err + if errors.As(err, &httpErr) && httpErr.Response != nil { + codes = append(codes, httpErr.StatusCode()) + } + + // plumbing.UnexpectedError does not implement Unwrap. + var unexpected *plumbing.UnexpectedError + if errors.As(err, &unexpected) && unexpected.Err != nil { + if errors.As(unexpected.Err, &httpErr) && httpErr.Response != nil { + codes = append(codes, httpErr.StatusCode()) } + } - start, end := from+i, from+i+len(code) - if (start == 0 || !isDigit(msg[start-1])) && (end == len(msg) || !isDigit(msg[end])) { - return true + for _, re := range []*regexp.Regexp{statusPhraseRE, statusPunctRE, statusTextRE} { + for _, m := range re.FindAllStringSubmatch(msg, -1) { + if n, convErr := strconv.Atoi(m[1]); convErr == nil { + codes = append(codes, n) + } } + } - from = start + 1 + return codes +} + +func hasAnyStatus(codes []int, match func(int) bool) bool { + for _, c := range codes { + if match(c) { + return true + } } + + return false } -func isDigit(b byte) bool { return b >= '0' && b <= '9' } +func isAuthStatus(c int) bool { return c == 401 || c == 403 } + +func isRetryableStatus(c int) bool { return c == 429 || c >= 500 && c <= 599 } func isAuthFailure(msg string, err error) bool { if strings.Contains(msg, "authentication required") || strings.Contains(msg, "invalid credentials") || strings.Contains(msg, "authorization failed") || strings.Contains(msg, "access denied") || - hasStatusCode(msg, "401") || - hasStatusCode(msg, "403 forbidden") { + hasAnyStatus(httpStatusCodes(err, msg), isAuthStatus) { return true } - if errors.Is(err, gittransport.ErrAuthenticationRequired) { + if errors.Is(err, gittransport.ErrAuthenticationRequired) || + errors.Is(err, gittransport.ErrAuthorizationFailed) { return true } diff --git a/internal/sink/git/errors_test.go b/internal/sink/git/errors_test.go index cc9ffd09..97fe3175 100644 --- a/internal/sink/git/errors_test.go +++ b/internal/sink/git/errors_test.go @@ -7,11 +7,15 @@ import ( "errors" "fmt" "io" + "net/http" + "net/url" "strings" "syscall" "testing" + "github.com/go-git/go-git/v5/plumbing" gittransport "github.com/go-git/go-git/v5/plumbing/transport" + githttp "github.com/go-git/go-git/v5/plumbing/transport/http" kollecterrors "github.com/platformrelay/kollect/internal/errors" ) @@ -153,27 +157,51 @@ func TestIsNonFastForwardError(t *testing.T) { } } -// Digits inside a path (a random temp dir) are not an HTTP status: a missing -// local remote must not classify as an auth failure. -func TestClassifyExportError_statusCodeNeedsTokenBoundary(t *testing.T) { +// Status codes are read from typed go-git errors and status phrases only; digit +// runs inside paths, hostnames and mirror dirs are not statuses. +func TestClassifyExportError_statusCodes(t *testing.T) { t.Parallel() - pathErr := errors.New("git clone: fatal: '/tmp/x3401029481/missing.git' does not appear to be a git repository") - if kollecterrors.IsTerminal(ClassifyExportError(pathErr)) { - t.Fatalf("digits in a path classified as terminal: %v", ClassifyExportError(pathErr)) - } - - for _, msg := range []string{"unexpected client error: 401", "HTTP 403 Forbidden"} { - if !kollecterrors.IsTerminal(ClassifyExportError(errors.New(msg))) { - t.Fatalf("%q must stay terminal", msg) + typed := plumbing.NewUnexpectedError(&githttp.Err{ + Response: &http.Response{StatusCode: 502, Request: &http.Request{URL: &url.URL{Scheme: "https", Host: "git.example.com", Path: "/team401/infra.git/info/refs"}}}, + }) + + rows := []struct { + name string + err error + want string // terminal | transient | other + }{ + {"missing local remote, digits in temp dir", errors.New("git clone: fatal: '/tmp/x3401029481/missing.git' does not appear to be a git repository"), "other"}, + {"typed 502 with team401 in URL", typed, "transient"}, + {"text 502 with team401 in URL", errors.New(`unexpected requesting "https://git.example.com/team401/infra.git/info/refs" status code: 502`), "transient"}, + {"hex mirror dir", errors.New("mirror dir: mkdir /var/git-mirrors/3fa401bc9e0d: permission denied"), "other"}, + {"hostname with 503", errors.New("lookup git-503.example.com: no such host"), "other"}, + {"digits inside 15039", errors.New("path /tmp/a15039b"), "other"}, + {"cli 403", errors.New("fatal: unable to access 'https://h/r.git/': The requested URL returned error: 403"), "terminal"}, + {"unexpected client error 401", errors.New("unexpected client error: 401"), "terminal"}, + {"http 403 forbidden", errors.New("HTTP 403 Forbidden"), "terminal"}, + {"status code 401", errors.New("status code: 401"), "terminal"}, + {"code=401", errors.New("code=401"), "terminal"}, + {"401 colon", errors.New("push: 401: bad"), "terminal"}, + {"remote rejected (403)", errors.New("remote rejected (403)"), "terminal"}, + {"wrapped authorization failed", fmt.Errorf("push: %w", gittransport.ErrAuthorizationFailed), "terminal"}, + {"returned error 503", errors.New("The requested URL returned error: 503"), "transient"}, + {"(429)", errors.New("rate limited (429)"), "transient"}, + {"status code 429", errors.New("status code: 429"), "transient"}, + } + + for _, row := range rows { + got := "other" + + switch classified := ClassifyExportError(row.err); { + case kollecterrors.IsTerminal(classified): + got = "terminal" + case strings.Contains(classified.Error(), "git transport"): + got = "transient" } - } - if !kollecterrors.IsTransient(ClassifyExportError(errors.New("server said 503"))) { - t.Fatal("standalone 503 must stay transient") - } - - if strings.Contains(ClassifyExportError(errors.New("path /tmp/a15039b")).Error(), "git transport") { - t.Fatal("503 inside a longer number must not be transient") + if got != row.want { + t.Errorf("%s: classified %s, want %s (%v)", row.name, got, row.want, row.err) + } } }