🐛 respawn cloudevents receiver after spontaneous Receive error

pkg/cloudevents/generic/clients/baseclient.go

@@ -292,7 +292,7 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 						receiverCtx, receiverCancel = context.WithCancel(ctx)
 						// Set flag before spawning goroutine to prevent race condition
 						startReceiving = true
-						go func() {
+						go func(receiverCtx context.Context) {
 							if err := c.transport.Receive(receiverCtx, func(handlerCtx context.Context, evt cloudevents.Event) {
 								receiveLogger := logging.SetLogTracingByCloudEvent(klog.FromContext(handlerCtx), &evt)
 								handlerCtx = klog.NewContext(handlerCtx, receiveLogger)
@@ -307,7 +307,44 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 							}); err != nil {
 								runtime.HandleErrorWithContext(ctx, err, "failed to receive cloudevents")
 							}
-						}()
+
+							// If the parent context is canceled, the client is shutting down.
+							if ctx.Err() != nil {
+								return
+							}
+							// If the receiver context was canceled by stopReceiverSignal, the
+							// connection monitor goroutine owns recovery (it will reconnect and
+							// re-trigger subscribeChan); nothing to do here.
+							if receiverCtx.Err() != nil {
+								return
+							}
+							// Receive returned spontaneously (e.g. transient gRPC Canceled /
+							// Unavailable on the inbound subscribe stream) while the parent
+							// context is still alive and the transport-level ErrorChan never
+							// fired. Without this branch, startReceiving stays true, no future
+							// startReceiverSignal can respawn the receiver, and bundle status
+							// updates are silently dropped until the pod restarts.
+							//
+							// Recover by sending stopReceiverSignal (clears startReceiving and
+							// cancels the now-dead receiver context) and re-triggering the
+							// subscribe goroutine, which will retry Subscribe with backoff and
+							// emit a fresh startReceiverSignal on success.
+							logger.V(2).Info("cloudevents receiver exited unexpectedly, triggering resubscribe")
+							select {
+							case c.receiverChan <- stopReceiverSignal:
+								// Signal sent successfully
+							default:
+								// Receiver channel is unavailable, that's okay - don't block
+								logger.V(2).Info("stopReceiverSignal not sent, receiver channel unavailable")
+							}
+							select {
+							case c.subscribeChan <- struct{}{}:
+								// Signal sent successfully
+							default:
+								// Subscribe channel is unavailable, that's okay - don't block
+								logger.V(2).Info("subscribe signal not sent, subscribe channel is unavailable")
+							}
+						}(receiverCtx)
 					}
 				case stopReceiverSignal:
 					logger.V(2).Info("stop the cloudevents receiver")

pkg/cloudevents/generic/clients/baseclient_test.go

@@ -0,0 +1,96 @@
+package clients
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/utils"
+)
+
+// recoveringTransport is a CloudEventTransport whose first Receive call returns
+// a transient error spontaneously (without firing ErrorChan), and whose
+// subsequent Receive calls block until the context is canceled. This models
+// the production failure mode where an inbound gRPC subscribe stream returns
+// e.g. codes.Canceled / codes.Unavailable while the transport-level connection
+// monitor never observes the failure.
+type recoveringTransport struct {
+	mu             sync.Mutex
+	errCh          chan error
+	receiveCalls   atomic.Int32
+	subscribeCalls atomic.Int32
+}
+
+func newRecoveringTransport() *recoveringTransport {
+	return &recoveringTransport{errCh: make(chan error, 1)}
+}
+
+func (t *recoveringTransport) Connect(ctx context.Context) error { return nil }
+func (t *recoveringTransport) Send(ctx context.Context, evt cloudevents.Event) error {
+	return nil
+}
+func (t *recoveringTransport) Subscribe(ctx context.Context) error {
+	t.subscribeCalls.Add(1)
+	return nil
+}
+
+func (t *recoveringTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
+	n := t.receiveCalls.Add(1)
+	if n == 1 {
+		// First invocation: return a transient error spontaneously.
+		// Mirrors the behavior of the gRPC transport when the inbound
+		// subscribe stream is canceled mid-flight without the client-side
+		// connection monitor detecting it via ErrorChan.
+		return fmt.Errorf("rpc error: code = Canceled desc = context canceled")
+	}
+	// Subsequent invocations: block until the receiver context is canceled.
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func (t *recoveringTransport) Close(ctx context.Context) error { return nil }
+func (t *recoveringTransport) ErrorChan() <-chan error         { return t.errCh }
+
+// TestReceiverRecoveryAfterSpontaneousReceiveError verifies that when
+// transport.Receive returns an error on its own (without the transport's
+// ErrorChan firing), the receiver lifecycle does not silently stay dead.
+// The baseClient must trigger a resubscribe so that a new Receive goroutine
+// is spawned and event processing resumes.
+//
+// Regression test for the production issue where ARO HCP's aro-hcp-backend
+// stopped processing maestro bundle status updates for hours because the
+// receiver goroutine launched inside (*baseClient).subscribe exited on a
+// transient gRPC Canceled error and was never restarted.
+func TestReceiverRecoveryAfterSpontaneousReceiveError(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	tr := newRecoveringTransport()
+	c := newBaseClient("test-client", tr, utils.EventRateLimit{})
+
+	if err := c.connect(ctx); err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+
+	c.subscribe(ctx, func(ctx context.Context, evt cloudevents.Event) {})
+
+	// Wait for at least two Receive invocations: the first returns the
+	// transient error, the second proves recovery (a new Receive goroutine
+	// was spawned via the resubscribe path).
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		if tr.receiveCalls.Load() >= 2 {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	t.Fatalf("receiver did not recover: receive=%d subscribe=%d (expected receive>=2)",
+		tr.receiveCalls.Load(), tr.subscribeCalls.Load())
+}