HYPERFLEET-1262 - fix: cap gRPC backoff and add retry for Maestro recovery

cmd/adapter/main.go

@@ -557,7 +557,7 @@ func runServe(flags *pflag.FlagSet) error {
 	}
 
 	// Create the event handler and subscribe to broker
-	handler := executor.AlwaysAck(executor.WithMetrics(exec.CreateHandler(), metricsRecorder, log))
+	handler := executor.AlwaysAck(executor.WithMetrics(exec.CreateHandler(), metricsRecorder, log), log)
 
 	// Handle signals for graceful shutdown
 	sigCh := make(chan os.Signal, 1)

go.mod

@@ -27,11 +27,12 @@ require (
 	go.opentelemetry.io/otel/sdk v1.44.0
 	go.opentelemetry.io/otel/trace v1.44.0
 	golang.org/x/text v0.39.0
+	google.golang.org/grpc v1.82.0
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/apimachinery v0.36.2
 	k8s.io/client-go v0.36.2
 	open-cluster-management.io/api v1.3.0
-	open-cluster-management.io/sdk-go v1.3.0
+	open-cluster-management.io/sdk-go v1.3.1-0.20260630085947-ac9666c85f0a
 	sigs.k8s.io/controller-runtime v0.24.1
 	sigs.k8s.io/yaml v1.6.0
 )
@@ -172,7 +173,6 @@ require (
 	google.golang.org/genproto v0.0.0-20260630182238-925bb5da69e7 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 // indirect
-	google.golang.org/grpc v1.82.0 // indirect
 	google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect

go.sum

@@ -519,8 +519,8 @@ k8s.io/utils v0.0.0-20260626114624-be93311217bd h1:Ea7fgQ5we8Y9T0OX5o0dAHzQOBRI0
 k8s.io/utils v0.0.0-20260626114624-be93311217bd/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk=
 open-cluster-management.io/api v1.3.0 h1:Q3miH38BE3N5+PesHQ0kcFi5nhX5350m7OJWapZcVqY=
 open-cluster-management.io/api v1.3.0/go.mod h1:t0DsBv4gjIo9ojd7GYfA2tcEMpNf0h5Ix68pFDXwNSk=
-open-cluster-management.io/sdk-go v1.3.0 h1:03VnnN3c10FwWnsSl9+apevKadLUw1hrKIu91JHL364=
-open-cluster-management.io/sdk-go v1.3.0/go.mod h1:+zL1hpfT73F4kxq3k293H6qsgm8k4YiC1giSwn36fAo=
+open-cluster-management.io/sdk-go v1.3.1-0.20260630085947-ac9666c85f0a h1:wdFYMV/zpSlapCadoyALzAS+mLslKJtdhe8M0i3n1kQ=
+open-cluster-management.io/sdk-go v1.3.1-0.20260630085947-ac9666c85f0a/go.mod h1:iPACd3YK3e4UMCoNzMDSdYGwKgLEPjM0VYl2nlIpE58=
 pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk=
 pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
 sigs.k8s.io/controller-runtime v0.24.1 h1:miPEwrmirImAvgME1L9qebGHrOnGJoVmVdtOU9fRfo4=

internal/executor/executor_test.go

@@ -925,7 +925,7 @@ func TestCreateHandler_MetricsRecording(t *testing.T) {
 				Build()
 			require.NoError(t, err)
 
-			handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger()))
+			handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger()), logger.NewTestLogger())
 
 			evt := event.New()
 			evt.SetID("test-event-1")
@@ -974,7 +974,7 @@ func TestCreateHandler_MetricsRecording_Failed(t *testing.T) {
 		Build()
 	require.NoError(t, err)
 
-	handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger()))
+	handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger()), logger.NewTestLogger())
 
 	evt := event.New()
 	evt.SetID("test-event-fail")
@@ -1013,7 +1013,7 @@ func TestCreateHandler_NilMetricsRecorder(t *testing.T) {
 		Build()
 	require.NoError(t, err)
 
-	handler := AlwaysAck(WithMetrics(exec.CreateHandler(), nil, logger.NewTestLogger()))
+	handler := AlwaysAck(WithMetrics(exec.CreateHandler(), nil, logger.NewTestLogger()), logger.NewTestLogger())
 
 	evt := event.New()
 	evt.SetID("test-event-nil")
@@ -1172,7 +1172,7 @@ func TestAlwaysAck_AlwaysReturnsNil(t *testing.T) {
 				return tt.result, tt.err
 			})
 
-			handler := AlwaysAck(inner)
+			handler := AlwaysAck(inner, logger.NewTestLogger())
 
 			evt := event.New()
 			evt.SetID("test-ack")

internal/executor/handler.go

@@ -11,7 +11,7 @@ import (
 
 // HandlerFunc is a composable event handler. Build a broker-compatible handler with:
 //
-//	handler := AlwaysAck(WithMetrics(exec.CreateHandler(), metricsRecorder, log))
+//	handler := AlwaysAck(WithMetrics(exec.CreateHandler(), metricsRecorder, log), log)
 type HandlerFunc func(ctx context.Context, evt *event.Event) (*ExecutionResult, error)
 
 // WithMetrics wraps a HandlerFunc to record Prometheus metrics after execution.
@@ -46,10 +46,25 @@ func WithMetrics(h HandlerFunc, recorder *metrics.Recorder, log logger.Logger) H
 
 // AlwaysAck wraps a HandlerFunc into a broker compatible handler that always returns nil,
 // preventing infinite retry loops for non-recoverable errors.
-// Any error returned by HandlerFunc is discarded, callers must log or record errors before this layer.
-func AlwaysAck(h HandlerFunc) func(ctx context.Context, evt *event.Event) error {
+// Errors are logged at warn level before being discarded.
+func AlwaysAck(h HandlerFunc, log logger.Logger) func(ctx context.Context, evt *event.Event) error {
 	return func(ctx context.Context, evt *event.Event) error {
-		_, _ = h(ctx, evt) //nolint:errcheck // errors are handled by inner wrappers before this layer
+		result, err := h(ctx, evt)
+		errCtx := logger.WithLogFields(ctx, logger.LogFields{
+			"event_id":   evt.ID(),
+			"event_type": evt.Type(),
+		})
+		if err != nil {
+			errCtx = logger.WithErrorField(errCtx, err)
+			log.Warn(errCtx, "event handler error (acked)")
+		} else if result != nil && result.Status == StatusFailed {
+			phases := make([]string, 0, len(result.Errors))
+			for phase := range result.Errors {
+				phases = append(phases, string(phase))
+			}
+			errCtx = logger.WithLogField(errCtx, "failed_phases", phases)
+			log.Warn(errCtx, "event handler failed (acked)")
+		}
 		return nil
 	}
 }

internal/maestroclient/client.go

@@ -21,6 +21,8 @@ import (
 	"github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/version"
 	"github.com/openshift-online/maestro/pkg/api/openapi"
 	"github.com/openshift-online/maestro/pkg/client/cloudevents/grpcsource"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/backoff"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -35,6 +37,7 @@ import (
 const (
 	DefaultHTTPTimeout              = 10 * time.Second
 	DefaultServerHealthinessTimeout = 20 * time.Second
+	DefaultGRPCBackoffMaxDelay      = 5 * time.Second
 )
 
 // Client is the Maestro client for managing ManifestWorks via CloudEvents gRPC
@@ -187,7 +190,19 @@ func NewMaestroClient(ctx context.Context, config *Config, log logger.Logger) (*
 
 	// Create gRPC options
 	grpcOptions := &grpcopts.GRPCOptions{
-		Dialer:                   &grpcopts.GRPCDialer{},
+		Dialer: &grpcopts.GRPCDialer{
+			ExtraDialOpts: []grpc.DialOption{
+				grpc.WithConnectParams(grpc.ConnectParams{
+					Backoff: backoff.Config{
+						BaseDelay:  1 * time.Second,
+						Multiplier: 1.6,
+						Jitter:     0.2,
+						MaxDelay:   DefaultGRPCBackoffMaxDelay,
+					},
+					MinConnectTimeout: 3 * time.Second,
+				}),
+			},
+		},
 		ServerHealthinessTimeout: &serverHealthinessTimeout,
 	}
 	grpcOptions.Dialer.URL = config.GRPCServerAddr

internal/maestroclient/operations.go

@@ -4,20 +4,62 @@ import (
 	"context"
 	"encoding/json"
 	"strings"
+	"time"
 
 	"github.com/openshift-hyperfleet/hyperfleet-adapter/internal/manifest"
 	"github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/constants"
 	apperrors "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/errors"
 	"github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/logger"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	kubetypes "k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	workv1 "open-cluster-management.io/api/work/v1"
 )
 
+const (
+	grpcRetryMaxAttempts = 3
+	grpcRetryBaseDelay   = 1 * time.Second
+	grpcRetryMultiplier  = 2
+)
+
+func isTransientGRPCError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return status.Code(err) == codes.Unavailable
+}
+
+func (c *Client) retryOnTransientGRPC(ctx context.Context, fn func() error) error {
+	var lastErr error
+	backoff := wait.Backoff{
+		Duration: grpcRetryBaseDelay,
+		Factor:   float64(grpcRetryMultiplier),
+		Steps:    grpcRetryMaxAttempts,
+	}
+	waitErr := wait.ExponentialBackoffWithContext(ctx, backoff, func(ctx context.Context) (bool, error) {
+		lastErr = fn()
+		if lastErr == nil {
+			return true, nil
+		}
+		if !isTransientGRPCError(lastErr) {
+			return false, lastErr
+		}
+		retryCtx := logger.WithErrorField(ctx, lastErr)
+		c.log.Warn(retryCtx, "transient gRPC error, retrying")
+		return false, nil
+	})
+	if wait.Interrupted(waitErr) && lastErr != nil {
+		return lastErr
+	}
+	return waitErr
+}
+
 // CreateManifestWork creates a new ManifestWork for a target cluster (consumer)
 //
 // The ManifestWork object should be pre-constructed from a template with:
@@ -54,8 +96,13 @@ func (c *Client) CreateManifestWork(
 	// Set namespace to consumer name (required by Maestro)
 	work.Namespace = consumerName
 
-	// Create via the work client
-	created, err := c.workClient.ManifestWorks(consumerName).Create(ctx, work, metav1.CreateOptions{})
+	// Create via the work client with retry on transient gRPC errors
+	var created *workv1.ManifestWork
+	err := c.retryOnTransientGRPC(ctx, func() error {
+		var createErr error
+		created, createErr = c.workClient.ManifestWorks(consumerName).Create(ctx, work, metav1.CreateOptions{})
+		return createErr
+	})
 	if err != nil {
 		if isConsumerNotFoundError(err) {
 			return nil, apperrors.NotFound("consumer %q is not registered in Maestro", consumerName)
@@ -76,9 +123,13 @@ func (c *Client) GetManifestWork(
 	ctx = logger.WithMaestroConsumer(ctx, consumerName)
 	ctx = logger.WithLogField(ctx, "manifestwork", workName)
 
-	work, err := c.workClient.ManifestWorks(consumerName).Get(ctx, workName, metav1.GetOptions{})
+	var work *workv1.ManifestWork
+	err := c.retryOnTransientGRPC(ctx, func() error {
+		var getErr error
+		work, getErr = c.workClient.ManifestWorks(consumerName).Get(ctx, workName, metav1.GetOptions{})
+		return getErr
+	})
 	if err != nil {
-		// Return not found error without wrapping for callers to check
 		if apierrors.IsNotFound(err) {
 			return nil, err
 		}
@@ -99,13 +150,18 @@ func (c *Client) PatchManifestWork(
 	ctx = logger.WithMaestroConsumer(ctx, consumerName)
 	ctx = logger.WithLogField(ctx, "manifestwork", workName)
 
-	patched, err := c.workClient.ManifestWorks(consumerName).Patch(
-		ctx,
-		workName,
-		kubetypes.MergePatchType,
-		patchData,
-		metav1.PatchOptions{},
-	)
+	var patched *workv1.ManifestWork
+	err := c.retryOnTransientGRPC(ctx, func() error {
+		var patchErr error
+		patched, patchErr = c.workClient.ManifestWorks(consumerName).Patch(
+			ctx,
+			workName,
+			kubetypes.MergePatchType,
+			patchData,
+			metav1.PatchOptions{},
+		)
+		return patchErr
+	})
 	if err != nil {
 		return nil, apperrors.MaestroError("failed to patch ManifestWork %s/%s: %v",
 			consumerName, workName, err)