🐛 fix(git): classify HTTP status codes from typed errors and status phrases; stabilise delete-watch envtest

internal/controller/family_sink_delete_watch_envtest_test.go

@@ -9,13 +9,15 @@ 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"
 
 	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,37 +48,60 @@ 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())
-
-		// 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
+		}
+
+		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))
+		}
+		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())
 

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,22 +68,77 @@ 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") ||
+		hasAnyStatus(httpStatusCodes(err, msg), isRetryableStatus) ||
 		strings.Contains(msg, "too many requests")
 }
 
+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())
+		}
+	}
+
+	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)
+			}
+		}
+	}
+
+	return codes
+}
+
+func hasAnyStatus(codes []int, match func(int) bool) bool {
+	for _, c := range codes {
+		if match(c) {
+			return true
+		}
+	}
+
+	return false
+}
+
+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") ||
-		strings.Contains(msg, "401") ||
-		strings.Contains(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
 	}
 

internal/sink/git/errors_test.go

@@ -7,10 +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"
 )
@@ -151,3 +156,52 @@ func TestIsNonFastForwardError(t *testing.T) {
 		t.Fatal("expected non-fast-forward for combined rejected-refs message")
 	}
 }
+
+// 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()
+
+	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 got != row.want {
+			t.Errorf("%s: classified %s, want %s (%v)", row.name, got, row.want, row.err)
+		}
+	}
+}