send the header immediately

Signed-off-by: Wei Liu <liuweixa@redhat.com>

Author: skeeey

pkg/cloudevents/generic/clients/baseclient.go

@@ -210,10 +210,23 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 			case <-ctx.Done():
 				return
 			case <-c.subscribeChan:
-				if err := c.transport.Subscribe(ctx); err != nil {
-					// Failed to send subscribe request, it should be connection failed, will retry on next reconnection
-					runtime.HandleErrorWithContext(ctx, err, "failed to subscribe after connection")
-					continue
+				// Retry subscribe with backoff until success or context cancellation
+				for {
+					if err := c.transport.Subscribe(ctx); err != nil {
+						runtime.HandleErrorWithContext(ctx, err, "failed to subscribe after connection")
+
+						// Wait with backoff before retrying
+						select {
+						case <-ctx.Done():
+							return
+						case <-wait.RealTimer(DelayFn()).C():
+							// Continue to retry
+						}
+						continue
+					}
+
+					// Subscribe succeeded, break out of retry loop
+					break
 				}
 
 				// Send startReceiverSignal to start/restart the receiver after successful subscription.

pkg/cloudevents/generic/options/v2/grpc/transport.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -120,7 +121,28 @@ func (t *grpcTransport) Subscribe(ctx context.Context) error {
 
 	values := header.Get(constants.GRPCSubscriptionIDKey)
 	if len(values) != 1 {
-		return fmt.Errorf("expected exactly one subscription-id header, got %d", len(values))
+		// Header() succeeded but no subscription-id was sent (header is nil or empty).
+		// This typically means the server rejected the subscription before sending headers
+		// (e.g., authorization failure). The actual error is only available via Recv().
+		// Call Recv() to get the real error from the server.
+		recvErrCh := make(chan error, 1)
+		go func() {
+			_, err := subClient.Recv()
+			recvErrCh <- err
+		}()
+		select {
+		case recvErr := <-recvErrCh:
+			if recvErr != nil {
+				return recvErr
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(5 * time.Second):
+			_ = subClient.CloseSend()
+			return fmt.Errorf("no subscription-id in header (%v): recv timeout", header)
+		}
+		// If Recv() didn't return an error, this is a server-side configuration issue
+		return fmt.Errorf("no subscription-id in header (%v)", header)
 	}
 	t.subID = values[0]
 	t.subClient = subClient

pkg/cloudevents/server/grpc/broker.go

@@ -124,34 +124,28 @@ func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest)
 	return &emptypb.Empty{}, nil
 }
 
-// register registers a subscriber and return client id and error channel.
-func (bkr *GRPCBroker) register(ctx context.Context,
+// registerSubscriber registers a subscriber with a pre-generated ID.
+// The subscription header must already be sent before calling this function.
+func (bkr *GRPCBroker) registerSubscriber(ctx context.Context,
+	id string,
 	dataType types.CloudEventsDataType,
 	subReq *pbv1.SubscriptionRequest,
-	subServer pbv1.CloudEventService_SubscribeServer,
-	handler resourceHandler) (string, error) {
+	handler resourceHandler) error {
 	logger := klog.FromContext(ctx)
 
 	bkr.mu.Lock()
 	defer bkr.mu.Unlock()
 
-	id := uuid.NewString()
+	logger.Info("registering subscriber", "id", id, "clusterName", subReq.ClusterName, "dataType", dataType)
+
 	bkr.subscribers[id] = &subscriber{
 		clusterName: subReq.ClusterName,
 		dataType:    dataType,
 		handler:     handler,
 	}
 
-	// Signal subscriber is registered
-	if err := subServer.SendHeader(metadata.Pairs(constants.GRPCSubscriptionIDKey, id)); err != nil {
-		logger.Error(err, "failed to send subscription header, unregister subscriber", "subID", id)
-		delete(bkr.subscribers, id)
-		return "", err
-	}
-	logger.V(4).Info("register a subscriber", "id", id, "clusterName", subReq.ClusterName, "dataType", dataType)
 	metrics.IncGRPCCESubscribersMetric(subReq.ClusterName, dataType.String())
-
-	return id, nil
+	return nil
 }
 
 // unregister a subscriber by id
@@ -181,10 +175,17 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		return fmt.Errorf("invalid subscription request: invalid data type %v", err)
 	}
 
+	// Generate subscription ID and send header IMMEDIATELY, before any other operations
+	// This ensures the client receives the header as soon as possible after the stream is established
+	subID := uuid.NewString()
+	if err := subServer.SendHeader(metadata.Pairs(constants.GRPCSubscriptionIDKey, subID)); err != nil {
+		return fmt.Errorf("failed to send subscription header for subID %s: %w", subID, err)
+	}
+
 	subCtx, cancel := context.WithCancel(subServer.Context())
 	defer cancel()
 
-	logger := klog.FromContext(subCtx).WithValues("clusterName", subReq.ClusterName)
+	logger := klog.FromContext(subCtx).WithValues("clusterName", subReq.ClusterName, "subID", subID)
 
 	// TODO make the channel size configurable
 	eventCh := make(chan *pbv1.CloudEvent, 100)
@@ -195,6 +196,35 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 	}
 	sendErrCh := make(chan error, 1)
 
+	// Register the subscriber with the ID we already created and sent in the header
+	err = bkr.registerSubscriber(klog.NewContext(subCtx, logger), subID, *dataType, subReq, func(handlerCtx context.Context, subID string, evt *cloudevents.Event) error {
+		// convert the cloudevents.Event to pbv1.CloudEvent
+		// WARNING: don't use "pbEvt, err := pb.ToProto(evt)" to convert cloudevent to protobuf
+		pbEvt := &pbv1.CloudEvent{}
+		if err := grpcprotocol.WritePBMessage(handlerCtx, binding.ToMessage(evt), pbEvt); err != nil {
+			return fmt.Errorf("failed to convert cloudevent to protobuf for resource(%s): %v", evt.ID(), err)
+		}
+
+		// send the cloudevent to the subscriber
+		klog.FromContext(handlerCtx).V(4).Info("sending the event to spec subscribers",
+			"subID", subID, "eventType", evt.Type(), "extensions", evt.Extensions())
+		select {
+		case eventCh <- pbEvt:
+		case <-subCtx.Done():
+			// The context of the stream has been canceled or completed.
+			// This could happen if:
+			// - The client closed the connection or canceled the stream.
+			// - The server closed the stream, potentially due to a shutdown.
+			// No error is returned here because the stream closure is expected.
+			return nil
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
 	// send events
 	// The grpc send is not concurrency safe and non-blocking, see: https://github.com/grpc/grpc-go/blob/v1.75.1/stream.go#L1571
 	// Return the error without wrapping, as it includes the gRPC error code and message for further handling.
@@ -238,34 +268,6 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		}
 	}()
 
-	subID, err := bkr.register(klog.NewContext(subCtx, logger), *dataType, subReq, subServer, func(handlerCtx context.Context, subID string, evt *cloudevents.Event) error {
-		// convert the cloudevents.Event to pbv1.CloudEvent
-		// WARNING: don't use "pbEvt, err := pb.ToProto(evt)" to convert cloudevent to protobuf
-		pbEvt := &pbv1.CloudEvent{}
-		if err := grpcprotocol.WritePBMessage(handlerCtx, binding.ToMessage(evt), pbEvt); err != nil {
-			return fmt.Errorf("failed to convert cloudevent to protobuf for resource(%s): %v", evt.ID(), err)
-		}
-
-		// send the cloudevent to the subscriber
-		logger.V(4).Info("sending the event to spec subscribers",
-			"subID", subID, "eventType", evt.Type(), "extensions", evt.Extensions())
-		select {
-		case eventCh <- pbEvt:
-		case <-subCtx.Done():
-			// The context of the stream has been canceled or completed.
-			// This could happen if:
-			// - The client closed the connection or canceled the stream.
-			// - The server closed the stream, potentially due to a shutdown.
-			// No error is returned here because the stream closure is expected.
-			return nil
-		}
-
-		return nil
-	})
-	if err != nil {
-		return err
-	}
-
 	if heartbeater != nil {
 		go heartbeater.Start(subCtx)
 	}