fix: prevent GCP Pub/Sub SDK receipt modack from overriding visibility timeout (#869)

* fix: prevent GCP Pub/Sub SDK receipt modack from overriding visibility timeout

The GCP Pub/Sub native SDK sends a ModifyAckDeadline ("receipt modack")
when it first receives a message from StreamingPull. The deadline value
is based on the SDK's internal p99 ack-latency distribution, with a
minimum of 10s. This overrides the subscription's ackDeadlineSeconds
on the server side.

With MaxExtension=-1 (which disables keep-alive extensions), the
effective visibility timeout becomes 10s instead of the configured 60s.
Any handler that takes >10s triggers premature Pub/Sub redelivery,
causing duplicate processing and DLQ entries.

Add VisibilityTimeout to QueueConfig and set MinExtensionPeriod to
match it in the GCP Pub/Sub subscription, preventing the receipt modack
from shortening the deadline below the configured value.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: shut down subscription in init test to prevent message stealing

The first sub-test left its subscription running, causing it to steal
messages published by the second sub-test. The stolen message sat until
the ack deadline expired and Pub/Sub redelivered it — 60s with our
MinExtensionPeriod fix, 10s on main. Test now runs in ~1s.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: alexluong

internal/config/mqconfig_aws.go

@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/hookdeck/outpost/internal/mqinfra"
 	"github.com/hookdeck/outpost/internal/mqs"
@@ -51,6 +52,7 @@ func (c *AWSSQSConfig) ToQueueConfig(ctx context.Context, queueType string) (*mq
 			ServiceAccountCredentials: c.getCredentials(),
 			Topic:                     c.getQueueName(queueType),
 		},
+		VisibilityTimeout: 60 * time.Second,
 	}, nil
 }
 

internal/config/mqconfig_azure.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"context"
+	"time"
 
 	"github.com/hookdeck/outpost/internal/mqinfra"
 	"github.com/hookdeck/outpost/internal/mqs"
@@ -96,6 +97,7 @@ func (c *AzureServiceBusConfig) ToQueueConfig(ctx context.Context, queueType str
 				Topic:            topic,
 				Subscription:     subscription,
 			},
+			VisibilityTimeout: 60 * time.Second,
 		}, nil
 	}
 
@@ -110,6 +112,7 @@ func (c *AzureServiceBusConfig) ToQueueConfig(ctx context.Context, queueType str
 			ResourceGroup:  c.ResourceGroup,
 			Namespace:      c.Namespace,
 		},
+		VisibilityTimeout: 60 * time.Second,
 	}, nil
 }
 

internal/config/mqconfig_gcp.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"context"
+	"time"
 
 	"github.com/hookdeck/outpost/internal/mqinfra"
 	"github.com/hookdeck/outpost/internal/mqs"
@@ -57,6 +58,7 @@ func (c *GCPPubSubConfig) ToQueueConfig(ctx context.Context, queueType string) (
 			TopicID:                   c.getTopicByQueueType(queueType),
 			SubscriptionID:            c.getSubscriptionByQueueType(queueType),
 		},
+		VisibilityTimeout: 60 * time.Second,
 	}, nil
 }
 

internal/config/mqconfig_rabbitmq.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"context"
+	"time"
 
 	"github.com/hookdeck/outpost/internal/mqinfra"
 	"github.com/hookdeck/outpost/internal/mqs"
@@ -42,6 +43,7 @@ func (c *RabbitMQConfig) ToQueueConfig(ctx context.Context, queueType string) (*
 			Exchange:  c.Exchange,
 			Queue:     c.getQueueName(queueType),
 		},
+		VisibilityTimeout: 60 * time.Second,
 	}, nil
 }
 

internal/mqs/queue.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"time"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
@@ -17,6 +18,8 @@ type QueueConfig struct {
 	GCPPubSub       *GCPPubSubConfig
 	RabbitMQ        *RabbitMQConfig
 	InMemory        *InMemoryConfig // mainly for testing purposes
+
+	VisibilityTimeout time.Duration
 }
 
 type InMemoryConfig struct {
@@ -88,7 +91,7 @@ func NewQueue(config *QueueConfig) Queue {
 	} else if config.AzureServiceBus != nil {
 		return NewAzureServiceBusQueue(config.AzureServiceBus)
 	} else if config.GCPPubSub != nil {
-		return NewGCPPubSubQueue(config.GCPPubSub)
+		return NewGCPPubSubQueue(config.GCPPubSub, config.VisibilityTimeout)
 	} else if config.RabbitMQ != nil {
 		return NewRabbitMQQueue(config.RabbitMQ)
 	} else {

internal/mqs/queue_gcppubsub.go

@@ -23,22 +23,24 @@ type GCPPubSubConfig struct {
 }
 
 type GCPPubSubQueue struct {
-	once       *sync.Once
-	base       *wrappedBaseQueue
-	config     *GCPPubSubConfig
-	topic      *pubsub.Topic
-	cleanupFns []func()
+	once              *sync.Once
+	base              *wrappedBaseQueue
+	config            *GCPPubSubConfig
+	visibilityTimeout time.Duration
+	topic             *pubsub.Topic
+	cleanupFns        []func()
 }
 
 var _ Queue = &GCPPubSubQueue{}
 
-func NewGCPPubSubQueue(config *GCPPubSubConfig) *GCPPubSubQueue {
+func NewGCPPubSubQueue(config *GCPPubSubConfig, visibilityTimeout time.Duration) *GCPPubSubQueue {
 	var once sync.Once
 	return &GCPPubSubQueue{
-		config:     config,
-		once:       &once,
-		base:       newWrappedBaseQueue(),
-		cleanupFns: []func(){},
+		config:            config,
+		visibilityTimeout: visibilityTimeout,
+		once:              &once,
+		base:              newWrappedBaseQueue(),
+		cleanupFns:        []func(){},
 	}
 }
 
@@ -152,6 +154,14 @@ func (q *GCPPubSubQueue) Subscribe(ctx context.Context, opts ...SubscribeOption)
 	// logic and do not want the SDK silently extending message leases — if a
 	// handler exceeds the ack deadline, the message should be redelivered.
 	sub.ReceiveSettings.MaxExtension = -1 * time.Second
+	// The native SDK sends a "receipt modack" (ModifyAckDeadline) when it first
+	// receives a message, using its internal ack-latency p99 as the deadline
+	// (minimum 10s). This overrides the subscription's ackDeadlineSeconds on the
+	// server side. Without MinExtensionPeriod, a subscription configured with a
+	// 60s ack deadline effectively becomes 10s, causing premature redelivery for
+	// any handler that takes >10s. Setting MinExtensionPeriod to match the
+	// subscription's visibility timeout prevents this override.
+	sub.ReceiveSettings.MinExtensionPeriod = q.visibilityTimeout
 
 	msgChan := make(chan *Message, concurrency)
 	subCtx, cancel := context.WithCancel(ctx)

internal/mqs/queue_test.go

@@ -103,12 +103,12 @@ func testMQ(t *testing.T, makeConfig func() mqs.QueueConfig) {
 		defer cleanup()
 		subscription, err := queue.Subscribe(context.Background())
 		require.Nil(t, err)
+		defer subscription.Shutdown(context.Background())
 		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 		defer cancel()
 		msg, err := subscription.Receive(ctx)
 		assert.Nil(t, msg)
 		assert.Equal(t, err, context.DeadlineExceeded)
-		defer cleanup()
 	})
 
 	t.Run("should publish and receive message", func(t *testing.T) {