test: verify event consumer slow-handler backpressure

sdk/go/event_integration_test.go

@@ -149,6 +149,79 @@ func TestIntegrationEventConsumerHeartbeatAndLostAck(t *testing.T) {
 	}
 	t.Log("real 15-second heartbeat and committed ACK loss recovered; handler and delivery attempt each remained one")
 }
+func TestIntegrationEventConsumerSlowHandlerBackpressure(t *testing.T) {
+	api, _, _ := integrationWorkerClient(t)
+	ctx, stop := context.WithTimeout(context.Background(), 25*time.Second)
+	defer stop()
+	sub := integrationEventSubscription(t, ctx, api, 3)
+	_ = integrationCancelledEvent(t, ctx, api, sub)
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var calls atomic.Int64
+	consumer, err := NewEventConsumer(api, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, _ ResultEvent) error {
+		if calls.Add(1) == 1 {
+			close(started)
+			select {
+			case <-release:
+			case <-work.Done():
+				return work.Err()
+			}
+		}
+		return nil
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	run, cancel := context.WithCancel(ctx)
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- consumer.Run(run) }()
+	select {
+	case <-started:
+	case <-ctx.Done():
+		t.Fatal("first slow handler did not start")
+	case err := <-done:
+		t.Fatal("consumer stopped", err)
+	}
+	_ = integrationCancelledEvent(t, ctx, api, sub)
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		current, getErr := api.GetSubscription(ctx, sub.ID)
+		if getErr != nil {
+			t.Fatal(getErr)
+		}
+		if current.PendingCount == 2 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("slow handler did not retain second delivery: pending=%d", current.PendingCount)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	time.Sleep(150 * time.Millisecond)
+	if calls.Load() != 1 || consumer.Stats().Acknowledged != 0 {
+		t.Fatalf("capacity-one consumer started or acknowledged work while handler was blocked: calls=%d stats=%+v", calls.Load(), consumer.Stats())
+	}
+	close(release)
+	for consumer.Stats().Acknowledged < 2 {
+		select {
+		case <-ctx.Done():
+			t.Fatal("slow consumer did not drain both deliveries")
+		case err := <-done:
+			t.Fatal("consumer stopped", err)
+		case <-time.After(20 * time.Millisecond):
+		}
+	}
+	cancel()
+	if err := awaitConsumerStop(t, done); err != nil {
+		t.Fatal(err)
+	}
+	page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
+	if err != nil || len(page.Items) != 2 || page.Items[0].Attempts != 1 || page.Items[1].Attempts != 1 || calls.Load() != 2 {
+		t.Fatalf("slow-consumer backpressure or delivery accounting failed: err=%v deliveries=%+v calls=%d", err, page.Items, calls.Load())
+	}
+	t.Log("capacity-one slow handler held the second delivery until release; both durable deliveries were acknowledged once")
+}
 func TestIntegrationEventConsumerFailureRedeliversNaturally(t *testing.T) {
 	api, _, _ := integrationWorkerClient(t)
 	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)