fix: make demo e2e catalog queries resilient to transient failures

The generate-demos CI job fails ~45% of the time with jq exit status 5
on the ClusterCatalog Quickstart scenario. Several issues contribute:

- jq -s (slurp mode) buffers the entire operatorhubio FBC response in
  memory before processing, risking system errors on large catalogs
- catalog content queries run exactly once with no retry, so any
  transient port-forward or network hiccup fails the step immediately
- bash() does not attach stderr to ExitError, making failures opaque
- with CatalogdHA, kubectl port-forward to the service may connect to
  a non-leader pod that returns 404 (empty local cache) and sticks
  with it for the entire retry window

Remove jq slurp mode so each JSON object is processed in constant
memory, prefixing filters with 'objects' to skip non-object values in
the FBC stream. Wrap CatalogContainsSomePackages, PackageHasSomeChannels,
and PackageHasSomeBundles in waitFor for retry on transient errors. Add
curl --compressed to handle gzip-encoded responses and --fail with
pipefail to detect HTTP errors. Re-establish dead port-forwards via
liveness checks and reset port-forwards on query failure so retries
can reach the leader pod. Inject stderr into ExitError in bash() to
match k8sClient diagnostics. Log catalog query errors at V(0) so CI
timeout failures are diagnosable.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: pedjak

test/e2e/steps/demo_steps.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -41,6 +42,10 @@ func bash(ctx context.Context, script string) (string, error) {
 
 	if err != nil {
 		logger.V(1).Info("Failed to run", "command", script, "stderr", stderr, "error", err)
+		var exitErr *exec.ExitError
+		if errors.As(err, &exitErr) {
+			exitErr.Stderr = stderrBuf.Bytes()
+		}
 	}
 	logger.V(1).Info("Output", "command", script, "output", stdout)
 
@@ -81,7 +86,15 @@ func CatalogReportsConditionWithoutReason(ctx context.Context, catalogUserName,
 func ensureCatalogPortForward(ctx context.Context) (string, error) {
 	sc := scenarioCtx(ctx)
 	if sc.catalogAddr != "" {
-		return sc.catalogAddr, nil
+		if catalogPortForwardAlive(sc.catalogAddr) {
+			return sc.catalogAddr, nil
+		}
+		logger.V(1).Info("Catalog port-forward is dead, re-establishing", "addr", sc.catalogAddr)
+		if sc.catalogCleanup != nil {
+			sc.catalogCleanup()
+		}
+		sc.catalogAddr = ""
+		sc.catalogCleanup = nil
 	}
 
 	addr, cleanup, err := portForward(ctx, componentNamespaces["catalogd"], "service/catalogd-service", 443)
@@ -92,68 +105,104 @@ func ensureCatalogPortForward(ctx context.Context) (string, error) {
 	sc.catalogCleanup = cleanup
 
 	waitFor(ctx, func() bool {
-		client := &http.Client{
-			Timeout: 3 * time.Second,
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec
-				DialContext:     (&net.Dialer{Timeout: 2 * time.Second}).DialContext,
-			},
-		}
-		resp, err := client.Get(fmt.Sprintf("https://%s/", addr))
-		if err != nil {
-			return false
-		}
-		resp.Body.Close()
-		return true
+		return catalogPortForwardAlive(addr)
 	})
 	return addr, nil
 }
 
+func catalogPortForwardAlive(addr string) bool {
+	client := &http.Client{
+		Timeout: 3 * time.Second,
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec
+			DialContext:     (&net.Dialer{Timeout: 2 * time.Second}).DialContext,
+		},
+	}
+	resp, err := client.Get(fmt.Sprintf("https://%s/", addr))
+	if err != nil {
+		return false
+	}
+	resp.Body.Close()
+	return true
+}
+
+// resetCatalogPortForward tears down the cached port-forward so the next
+// call to ensureCatalogPortForward establishes a fresh connection.  With
+// CatalogdHA, non-leader pods return 404 (empty local cache); resetting
+// lets the next retry potentially reach the leader pod.
+func resetCatalogPortForward(ctx context.Context) {
+	sc := scenarioCtx(ctx)
+	if sc.catalogCleanup != nil {
+		sc.catalogCleanup()
+	}
+	sc.catalogAddr = ""
+	sc.catalogCleanup = nil
+}
+
 func catalogCurlJq(ctx context.Context, catalogName, jqFilter string) (string, error) {
 	addr, err := ensureCatalogPortForward(ctx)
 	if err != nil {
 		return "", err
 	}
 	script := fmt.Sprintf(
-		`curl -s -k https://%s/catalogs/%s/api/v1/all | jq -s '%s'`,
+		`set -o pipefail; curl -sS -k --compressed --fail https://%s/catalogs/%s/api/v1/all | jq '%s'`,
 		addr, catalogName, jqFilter,
 	)
-	return bash(ctx, script)
+	out, err := bash(ctx, script)
+	if err != nil {
+		resetCatalogPortForward(ctx)
+	}
+	return out, err
 }
 
 func CatalogContainsSomePackages(ctx context.Context, catalogName string) error {
-	out, err := catalogCurlJq(ctx, catalogName,
-		`.[] | select(.schema == "olm.package") | .name`)
-	if err != nil {
-		return err
-	}
-	if strings.TrimSpace(out) == "" {
-		return fmt.Errorf("catalog %q contains no packages", catalogName)
-	}
+	waitFor(ctx, func() bool {
+		out, err := catalogCurlJq(ctx, catalogName,
+			`objects | select(.schema == "olm.package") | .name`)
+		if err != nil {
+			logger.Info("Catalog query failed, retrying", "catalog", catalogName, "error", err, "stderr", stderrOutput(err))
+			return false
+		}
+		if strings.TrimSpace(out) == "" {
+			logger.Info("Catalog returned no packages, retrying", "catalog", catalogName)
+			return false
+		}
+		return true
+	})
 	return nil
 }
 
 func PackageHasSomeChannels(ctx context.Context, packageName, catalogName string) error {
-	out, err := catalogCurlJq(ctx, catalogName,
-		fmt.Sprintf(`.[] | select(.schema == "olm.channel") | select(.package == "%s") | .name`, packageName))
-	if err != nil {
-		return err
-	}
-	if strings.TrimSpace(out) == "" {
-		return fmt.Errorf("package %q in catalog %q has no channels", packageName, catalogName)
-	}
+	waitFor(ctx, func() bool {
+		out, err := catalogCurlJq(ctx, catalogName,
+			fmt.Sprintf(`objects | select(.schema == "olm.channel") | select(.package == "%s") | .name`, packageName))
+		if err != nil {
+			logger.Info("Catalog query failed, retrying", "catalog", catalogName, "package", packageName, "error", err, "stderr", stderrOutput(err))
+			return false
+		}
+		if strings.TrimSpace(out) == "" {
+			logger.Info("Package has no channels, retrying", "catalog", catalogName, "package", packageName)
+			return false
+		}
+		return true
+	})
 	return nil
 }
 
 func PackageHasSomeBundles(ctx context.Context, packageName, catalogName string) error {
-	out, err := catalogCurlJq(ctx, catalogName,
-		fmt.Sprintf(`.[] | select(.schema == "olm.bundle") | select(.package == "%s") | .name`, packageName))
-	if err != nil {
-		return err
-	}
-	if strings.TrimSpace(out) == "" {
-		return fmt.Errorf("package %q in catalog %q has no bundles", packageName, catalogName)
-	}
+	waitFor(ctx, func() bool {
+		out, err := catalogCurlJq(ctx, catalogName,
+			fmt.Sprintf(`objects | select(.schema == "olm.bundle") | select(.package == "%s") | .name`, packageName))
+		if err != nil {
+			logger.Info("Catalog query failed, retrying", "catalog", catalogName, "package", packageName, "error", err, "stderr", stderrOutput(err))
+			return false
+		}
+		if strings.TrimSpace(out) == "" {
+			logger.Info("Package has no bundles, retrying", "catalog", catalogName, "package", packageName)
+			return false
+		}
+		return true
+	})
 	return nil
 }