Function TestSubscribe_ErrorHandling

lib/gcppubsub/client_test.go:230–321

(t *testing.T)

}

func TestSubscribe_ErrorHandling(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	topicID := "error-handling-topic"
	subID := "error-handling-sub"

	// Setup Topic & Sub
	createTestTopic(t, topicID)
	createTestSubscription(t, topicID, subID, 1) // Short deadline for NACK retry test

	// Case 1: Transient Error (Should Retry)
	t.Run("TransientError_Retries", func(t *testing.T) {
		msgID, err := pubsubClient.Publish(ctx, topicID, []byte("retry-me"))
		if err != nil {
			// Should not error out of Publish
			t.Fatalf("Publish failed: %v", err)
		}

		var attempts atomic.Int32
		done := make(chan struct{})

		go func() {
			err := pubsubClient.Subscribe(ctx, subID, func(_ context.Context, id string, data []byte) error {
				if !isExpectedMessage(t, []byte("retry-me"), data, msgID, id) {
					return nil // Ignore other messages
				}

				count := attempts.Add(1)
				if count == 1 {
					// First attempt: Simulate Transient Error
					return event.ErrTransientFailure
				}
				// Second attempt: Success
				close(done)

				return nil
			})
			if err != nil {
				// Should not error out of Subscribe
				t.Errorf("Subscribe failed: %v", err)
			}
		}()

		select {
		case <-done:
			if attempts.Load() < 2 {
				t.Errorf("Expected retries, but succeeded on attempt %d", attempts.Load())
			}
		case <-ctx.Done():
			t.Fatal("Timeout waiting for retry")
		}
	})

	// Case 2: Permanent Error (Should ACK and NOT Retry)
	t.Run("PermanentError_NoRetry", func(t *testing.T) {
		// New subscription to isolate logic
		permSubID := "perm-error-sub"