send the header immediately

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

Author: skeeey

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

@@ -120,7 +120,16 @@ 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.
+		_, recvErr := subClient.Recv()
+		if recvErr != nil {
+			return recvErr
+		}
+		// If Recv() didn't return an error, this is a server-side configuration issue
+		return fmt.Errorf("no subscription-id in header (%v)", values)
 	}
 	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)
 	}

pkg/cloudevents/server/grpc/broker_test.go

@@ -133,3 +133,227 @@ func TestServer(t *testing.T) {
 		t.Error("received event is different")
 	}
 }
+
+// TestSubscriptionHeaderImmediateSend verifies that the subscription ID header
+// is sent immediately upon subscription, preventing the "got 0 headers" error.
+func TestSubscriptionHeaderImmediateSend(t *testing.T) {
+	grpcServerOptions := []grpc.ServerOption{}
+	grpcServer := grpc.NewServer(grpcServerOptions...)
+	defer grpcServer.Stop()
+
+	grpcEventServer := NewGRPCBroker(NewBrokerOptions())
+	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
+
+	svc := &testService{evts: make(map[string]*cloudevents.Event)}
+	grpcEventServer.RegisterService(context.Background(), dataType, svc)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	t.Cleanup(func() {
+		grpcServer.GracefulStop()
+		_ = lis.Close()
+	})
+
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			t.Errorf("failed to serve: %v", err)
+		}
+	}()
+
+	grpcClientOptions := grpccli.NewGRPCOptions()
+	grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
+	agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType)
+
+	// Test that connection and subscription work without "got 0 headers" error
+	if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+
+	if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil {
+		t.Fatalf("failed to subscribe, expected header to be sent immediately: %v", err)
+	}
+}
+
+// TestReconnectionScenario simulates a client restart (disconnect and reconnect)
+// to verify that subscription headers are properly sent on reconnection.
+func TestReconnectionScenario(t *testing.T) {
+	grpcServerOptions := []grpc.ServerOption{}
+	grpcServer := grpc.NewServer(grpcServerOptions...)
+	defer grpcServer.Stop()
+
+	grpcEventServer := NewGRPCBroker(NewBrokerOptions())
+	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
+
+	svc := &testService{evts: make(map[string]*cloudevents.Event)}
+	grpcEventServer.RegisterService(context.Background(), dataType, svc)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	t.Cleanup(func() {
+		grpcServer.GracefulStop()
+		_ = lis.Close()
+	})
+
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			t.Errorf("failed to serve: %v", err)
+		}
+	}()
+
+	grpcClientOptions := grpccli.NewGRPCOptions()
+	grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
+
+	// First connection and subscription
+	agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType)
+	if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+
+	if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil {
+		t.Fatalf("failed to subscribe on first connection: %v", err)
+	}
+
+	// Simulate client restart by closing and reconnecting
+	if err := agentOption.CloudEventsTransport.Close(ctx); err != nil {
+		t.Fatalf("failed to close: %v", err)
+	}
+
+	// Create a new transport for reconnection
+	grpcClientOptions2 := grpccli.NewGRPCOptions()
+	grpcClientOptions2.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
+	agentOption2 := grpcv2.NewAgentOptions(grpcClientOptions2, "cluster1", "agent1", dataType)
+
+	// Reconnect
+	if err := agentOption2.CloudEventsTransport.Connect(ctx); err != nil {
+		t.Fatalf("failed to reconnect: %v", err)
+	}
+
+	// This should not fail with "got 0 headers" error
+	if err := agentOption2.CloudEventsTransport.Subscribe(ctx); err != nil {
+		t.Fatalf("failed to subscribe after reconnection: %v", err)
+	}
+}
+
+// TestConcurrentSubscriptions verifies that multiple clients can subscribe
+// simultaneously without header race conditions.
+func TestConcurrentSubscriptions(t *testing.T) {
+	grpcServerOptions := []grpc.ServerOption{}
+	grpcServer := grpc.NewServer(grpcServerOptions...)
+	defer grpcServer.Stop()
+
+	grpcEventServer := NewGRPCBroker(NewBrokerOptions())
+	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
+
+	svc := &testService{evts: make(map[string]*cloudevents.Event)}
+	grpcEventServer.RegisterService(context.Background(), dataType, svc)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	t.Cleanup(func() {
+		grpcServer.GracefulStop()
+		_ = lis.Close()
+	})
+
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			t.Errorf("failed to serve: %v", err)
+		}
+	}()
+
+	// Create and subscribe multiple clients concurrently
+	numClients := 10
+	errCh := make(chan error, numClients)
+
+	for i := 0; i < numClients; i++ {
+		go func(clientID int) {
+			grpcClientOptions := grpccli.NewGRPCOptions()
+			grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
+			agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType)
+
+			if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil {
+				errCh <- err
+				return
+			}
+
+			if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil {
+				errCh <- err
+				return
+			}
+
+			errCh <- nil
+		}(i)
+	}
+
+	// Wait for all clients to complete
+	for i := 0; i < numClients; i++ {
+		if err := <-errCh; err != nil {
+			t.Errorf("client %d failed: %v", i, err)
+		}
+	}
+}
+
+// TestMultipleRapidReconnections simulates rapid reconnection scenarios
+// that could trigger the race condition.
+func TestMultipleRapidReconnections(t *testing.T) {
+	grpcServerOptions := []grpc.ServerOption{}
+	grpcServer := grpc.NewServer(grpcServerOptions...)
+	defer grpcServer.Stop()
+
+	grpcEventServer := NewGRPCBroker(NewBrokerOptions())
+	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
+
+	svc := &testService{evts: make(map[string]*cloudevents.Event)}
+	grpcEventServer.RegisterService(context.Background(), dataType, svc)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	t.Cleanup(func() {
+		grpcServer.GracefulStop()
+		_ = lis.Close()
+	})
+
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			t.Errorf("failed to serve: %v", err)
+		}
+	}()
+
+	// Perform multiple rapid reconnections
+	for i := 0; i < 5; i++ {
+		grpcClientOptions := grpccli.NewGRPCOptions()
+		grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
+		agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType)
+
+		if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil {
+			t.Fatalf("reconnection %d: failed to connect: %v", i, err)
+		}
+
+		if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil {
+			t.Fatalf("reconnection %d: failed to subscribe: %v", i, err)
+		}
+
+		if err := agentOption.CloudEventsTransport.Close(ctx); err != nil {
+			t.Fatalf("reconnection %d: failed to close: %v", i, err)
+		}
+	}
+}