perf(retrymq): cut idle Redis polling to one command per interval

internal/config/config.go

@@ -3,6 +3,7 @@ package config
 import (
 	"errors"
 	"fmt"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -80,7 +81,7 @@ type Config struct {
 	RetrySchedule                 []int `yaml:"retry_schedule" env:"RETRY_SCHEDULE" envSeparator:"," desc:"Comma-separated list of retry delays in seconds. If provided, overrides retry_interval_seconds and retry_max_limit. Schedule length defines the max number of retries. Example: '5,60,600,3600,7200' for 5 retries at 5s, 1m, 10m, 1h, 2h." required:"N"`
 	RetryIntervalSeconds          int   `yaml:"retry_interval_seconds" env:"RETRY_INTERVAL_SECONDS" desc:"Interval in seconds for exponential backoff retry strategy (base 2). Ignored if retry_schedule is provided." required:"N"`
 	RetryMaxLimit                 int   `yaml:"retry_max_limit" env:"MAX_RETRY_LIMIT" desc:"Maximum number of retry attempts for a single event delivery before giving up. Ignored if retry_schedule is provided." required:"N"`
-	RetryPollBackoffMs            int   `yaml:"retry_poll_backoff_ms" env:"RETRY_POLL_BACKOFF_MS" desc:"Backoff time in milliseconds when the retry monitor finds no messages to process. When a retry message is found, the monitor immediately polls for the next message without delay. Lower values provide faster retry processing but increase Redis load. For serverless Redis providers (Upstash, ElastiCache Serverless), consider increasing to 5000-10000ms to reduce costs. Default: 100" required:"N"`
+	RetryPollBackoffMs            int   `yaml:"retry_poll_backoff_ms" env:"RETRY_POLL_BACKOFF_MS" desc:"Maximum time in milliseconds the retry monitor waits between polls while idle. When a retry is scheduled but not yet due, the monitor instead waits until it comes due. 0 or unset means auto: sleep until the next due message, at most min(30s, shortest configured retry delay), so retries are never late and idle cost is about one Redis command per interval. An explicit positive value is honored as-is as a fixed maximum idle sleep, which may add up to that much latency for retries scheduled while the monitor sleeps. When a retry message is found, the monitor immediately polls for the next message without delay. Default: 0 (auto)" required:"N"`
 	RetryVisibilityTimeoutSeconds int   `yaml:"retry_visibility_timeout_seconds" env:"RETRY_VISIBILITY_TIMEOUT_SECONDS" desc:"Time in seconds a retry message is hidden after being received before becoming visible again for reprocessing. This applies when event data is temporarily unavailable (e.g., race condition with log persistence). Default: 30" required:"N"`
 
 	// Event Delivery
@@ -168,7 +169,7 @@ func (c *Config) InitDefaults() {
 	c.RetrySchedule = []int{} // Empty by default, falls back to exponential backoff
 	c.RetryIntervalSeconds = 30
 	c.RetryMaxLimit = 10
-	c.RetryPollBackoffMs = 100
+	c.RetryPollBackoffMs = 0 // 0 = auto: min(30s, shortest configured retry delay)
 	c.RetryVisibilityTimeoutSeconds = 30
 	c.MaxDestinationsPerTenant = 20
 	c.DeliveryTimeoutSeconds = 5
@@ -603,6 +604,30 @@ func (c *Config) GetRetryBackoff() (backoff.Backoff, int) {
 	}, c.RetryMaxLimit
 }
 
+// defaultRetryPollBackoff is the ceiling for the auto retry poll backoff
+// (RetryPollBackoffMs = 0): the monitor never sleeps longer than this while
+// idle, even when the shortest configured retry delay is longer.
+const defaultRetryPollBackoff = 30 * time.Second
+
+// GetRetryPollBackoff returns the maximum time the retry monitor waits between
+// polls while idle. An explicitly configured positive value is honored as-is.
+// Otherwise (0 = auto) it is min(defaultRetryPollBackoff, shortest configured
+// retry delay), so the monitor is always awake by the time the earliest
+// possible retry comes due and the idle interval never adds latency.
+func (c *Config) GetRetryPollBackoff() time.Duration {
+	if c.RetryPollBackoffMs > 0 {
+		return time.Duration(c.RetryPollBackoffMs) * time.Millisecond
+	}
+	shortest := time.Duration(c.RetryIntervalSeconds) * time.Second
+	if len(c.RetrySchedule) > 0 {
+		shortest = time.Duration(slices.Min(c.RetrySchedule)) * time.Second
+	}
+	if shortest > 0 && shortest < defaultRetryPollBackoff {
+		return shortest
+	}
+	return defaultRetryPollBackoff
+}
+
 type TelemetryConfig struct {
 	Disabled          bool   `yaml:"disabled" env:"DISABLE_TELEMETRY" desc:"Disables telemetry within the 'telemetry' block (Hookdeck usage stats and Sentry). Can be overridden by the global 'disable_telemetry' flag at the root of the configuration." required:"N"`
 	BatchSize         int    `yaml:"batch_size" env:"TELEMETRY_BATCH_SIZE" desc:"Maximum number of telemetry events to batch before sending." required:"N"`

internal/config/config_retry_test.go

@@ -2,6 +2,7 @@ package config_test
 
 import (
 	"testing"
+	"time"
 
 	"github.com/hookdeck/outpost/internal/config"
 	"github.com/stretchr/testify/assert"
@@ -105,3 +106,108 @@ retry_interval_seconds: 60
 		})
 	}
 }
+
+func TestGetRetryPollBackoff(t *testing.T) {
+	tests := []struct {
+		name string
+		yaml string
+		want time.Duration
+	}{
+		{
+			name: "default is auto, 30s ceiling",
+			yaml: "",
+			want: 30 * time.Second,
+		},
+		{
+			name: "auto follows a retry interval shorter than the ceiling",
+			yaml: "retry_interval_seconds: 5\n",
+			want: 5 * time.Second,
+		},
+		{
+			name: "auto follows the shortest entry of a custom schedule",
+			yaml: "retry_schedule: [10, 5, 300]\n",
+			want: 5 * time.Second,
+		},
+		{
+			name: "auto stays at the 30s ceiling under a longer schedule",
+			yaml: "retry_schedule: [60, 300]\n",
+			want: 30 * time.Second,
+		},
+		{
+			name: "an explicit backoff below the auto value is used as-is",
+			yaml: "retry_poll_backoff_ms: 100\n",
+			want: 100 * time.Millisecond,
+		},
+		{
+			name: "an explicit backoff longer than the shortest delay is honored, not capped",
+			yaml: "retry_schedule: [5]\nretry_poll_backoff_ms: 10000\n",
+			want: 10 * time.Second,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockOS := &mockOS{
+				files:   map[string][]byte{"config.yaml": []byte(tt.yaml)},
+				envVars: map[string]string{"CONFIG": "config.yaml"},
+			}
+
+			mockOS.envVars["API_KEY"] = "test-key"
+			mockOS.envVars["API_JWT_SECRET"] = "test-jwt-secret"
+			mockOS.envVars["AES_ENCRYPTION_SECRET"] = "test-aes-secret-16b"
+			mockOS.envVars["POSTGRES_URL"] = "postgres://localhost:5432/test"
+			mockOS.envVars["RABBITMQ_SERVER_URL"] = "amqp://localhost:5672"
+
+			cfg, err := config.ParseWithOS(config.Flags{}, mockOS)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.want, cfg.GetRetryPollBackoff())
+		})
+	}
+}
+
+func TestRetryConfigurationValidation(t *testing.T) {
+	tests := []struct {
+		name    string
+		yaml    string
+		wantErr string
+	}{
+		{
+			name:    "zero retry_schedule entry rejected",
+			yaml:    "retry_schedule: [5, 0, 300]\n",
+			wantErr: "retry_schedule entries must be at least 1 second",
+		},
+		{
+			name:    "negative retry_schedule entry rejected",
+			yaml:    "retry_schedule: [-5, 300]\n",
+			wantErr: "retry_schedule entries must be at least 1 second",
+		},
+		{
+			name:    "zero retry_interval_seconds rejected",
+			yaml:    "retry_interval_seconds: 0\n",
+			wantErr: "retry_interval_seconds must be at least 1",
+		},
+		{
+			name:    "negative retry_poll_backoff_ms rejected",
+			yaml:    "retry_poll_backoff_ms: -1\n",
+			wantErr: "retry_poll_backoff_ms must not be negative",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockOS := &mockOS{
+				files:   map[string][]byte{"config.yaml": []byte(tt.yaml)},
+				envVars: map[string]string{"CONFIG": "config.yaml"},
+			}
+
+			mockOS.envVars["API_KEY"] = "test-key"
+			mockOS.envVars["API_JWT_SECRET"] = "test-jwt-secret"
+			mockOS.envVars["AES_ENCRYPTION_SECRET"] = "test-aes-secret-16b"
+			mockOS.envVars["POSTGRES_URL"] = "postgres://localhost:5432/test"
+			mockOS.envVars["RABBITMQ_SERVER_URL"] = "amqp://localhost:5672"
+
+			_, err := config.ParseWithOS(config.Flags{}, mockOS)
+			assert.ErrorContains(t, err, tt.wantErr)
+		})
+	}
+}

internal/config/validation.go

@@ -192,6 +192,17 @@ func (c *Config) validateDeploymentID() error {
 
 // validateRetryConfiguration validates and adjusts the retry configuration
 func (c *Config) validateRetryConfiguration() error {
+	for i, seconds := range c.RetrySchedule {
+		if seconds < 1 {
+			return fmt.Errorf("config validation error: retry_schedule entries must be at least 1 second, got %d at index %d", seconds, i)
+		}
+	}
+	if len(c.RetrySchedule) == 0 && c.RetryIntervalSeconds < 1 {
+		return fmt.Errorf("config validation error: retry_interval_seconds must be at least 1, got %d", c.RetryIntervalSeconds)
+	}
+	if c.RetryPollBackoffMs < 0 {
+		return fmt.Errorf("config validation error: retry_poll_backoff_ms must not be negative, got %d", c.RetryPollBackoffMs)
+	}
 	// If retry_schedule is provided, override retry_max_limit to match schedule length
 	if len(c.RetrySchedule) > 0 {
 		c.RetryMaxLimit = len(c.RetrySchedule)

internal/rsmq/rsmq.go

@@ -142,10 +142,23 @@ type QueueMessage struct {
 	Sent    time.Time
 }
 
+// PollResult is the outcome of a ReceiveMessagePoll call.
+type PollResult struct {
+	// Message is the received message, or nil if nothing was due.
+	Message *QueueMessage
+	// HasNext reports whether the queue holds any message at all. It is only
+	// meaningful when Message is nil.
+	HasNext bool
+	// NextDue is how long until the earliest message in the queue becomes
+	// visible. It is only meaningful when Message is nil and HasNext is true,
+	// and it is never negative.
+	NextDue time.Duration
+}
+
 // Client is the subset of *RedisSMQ methods used by consumers (e.g., scheduler)
 type Client interface {
 	CreateQueue(qname string, vt uint, delay uint, maxsize int) error
-	ReceiveMessage(qname string, vt uint) (*QueueMessage, error)
+	ReceiveMessagePoll(qname string, vt uint) (PollResult, error)
 	SendMessage(qname string, message string, delay uint, opts ...SendMessageOption) (string, error)
 	ChangeMessageVisibility(qname string, id string, vt uint) error
 	DeleteMessage(qname string, id string) error
@@ -524,31 +537,80 @@ func (rsmq *RedisSMQ) SendMessage(qname string, message string, delay uint, opts
 
 // ReceiveMessage receives message from the queue
 func (rsmq *RedisSMQ) ReceiveMessage(qname string, vt uint) (*QueueMessage, error) {
-	if err := validateQname(qname); err != nil {
-		return nil, err
-	}
-
-	queue, err := rsmq.getQueue(qname, true)
+	res, err := rsmq.ReceiveMessagePoll(qname, vt)
 	if err != nil {
 		return nil, err
 	}
+	return res.Message, nil
+}
 
-	if vt == UnsetVt {
-		vt = queue.vt
+// ReceiveMessagePoll receives a message from the queue and, when nothing is
+// due, reports when the queue's earliest message becomes visible so the caller
+// can sleep until then instead of polling on a fixed interval.
+func (rsmq *RedisSMQ) ReceiveMessagePoll(qname string, vt uint) (PollResult, error) {
+	if err := validateQname(qname); err != nil {
+		return PollResult{}, err
 	}
 
-	if err := validateVt(vt); err != nil {
-		return nil, err
+	vtArg := ""
+	if vt != UnsetVt {
+		if err := validateVt(vt); err != nil {
+			return PollResult{}, err
+		}
+		vtArg = strconv.FormatUint(uint64(vt), 10)
 	}
 
 	key := rsmq.ns + qname
 	hashKey := key + q // key + ":Q"
 
-	qvt := strconv.FormatUint(queue.ts+uint64(vt)*1000, 10)
-	ct := strconv.FormatUint(queue.ts, 10)
+	evalCmd := rsmq.client.EvalSha(hashReceiveMessage, []string{key, hashKey}, vtArg)
+	return rsmq.createPollResult(evalCmd)
+}
 
-	evalCmd := rsmq.client.EvalSha(hashReceiveMessage, []string{key, hashKey}, ct, qvt)
-	return rsmq.createQueueMessage(evalCmd)
+func (rsmq *RedisSMQ) createPollResult(cmd *redis.Cmd) (PollResult, error) {
+	if err := cmd.Err(); err != nil {
+		return PollResult{}, fmt.Errorf("rsmq command failed: %w", err)
+	}
+
+	vals, err := toAnySlice(cmd.Val())
+	if err != nil {
+		return PollResult{}, err
+	}
+	if len(vals) == 0 {
+		return PollResult{}, fmt.Errorf("empty receiveMessage response")
+	}
+
+	tag, ok := vals[0].(string)
+	if !ok {
+		return PollResult{}, fmt.Errorf("mismatched receiveMessage tag type: got %T", vals[0])
+	}
+
+	switch tag {
+	case "n":
+		return PollResult{}, ErrQueueNotFound
+	case "e":
+		if len(vals) != 3 {
+			return PollResult{}, fmt.Errorf("malformed receiveMessage empty response: %v", vals)
+		}
+		if convertIntToUint(vals[1]) == 0 {
+			return PollResult{}, nil
+		}
+		return PollResult{
+			HasNext: true,
+			NextDue: time.Duration(convertIntToUint(vals[2])) * time.Millisecond,
+		}, nil
+	case "m":
+		if len(vals) != 5 {
+			return PollResult{}, fmt.Errorf("malformed receiveMessage response: %v", vals)
+		}
+		msg, err := buildQueueMessage(vals[1:])
+		if err != nil {
+			return PollResult{}, err
+		}
+		return PollResult{Message: msg}, nil
+	default:
+		return PollResult{}, fmt.Errorf("unknown receiveMessage tag: %q", tag)
+	}
 }
 
 // PopMessage pop message from queue
@@ -577,28 +639,34 @@ func (rsmq *RedisSMQ) createQueueMessage(cmd *redis.Cmd) (*QueueMessage, error)
 		return nil, fmt.Errorf("rsmq command failed: %w", err)
 	}
 
-	val := cmd.Val()
+	vals, err := toAnySlice(cmd.Val())
+	if err != nil {
+		return nil, err
+	}
+	if len(vals) == 0 {
+		return nil, nil
+	}
+	return buildQueueMessage(vals)
+}
 
-	// Handle nil response - some Redis-compatible databases (e.g., Dragonfly)
-	// may return nil instead of empty array when no message is available
+// toAnySlice normalizes a script reply into a slice. A nil reply becomes an
+// empty slice — some Redis-compatible databases (e.g., Dragonfly) return nil
+// instead of an empty array.
+func toAnySlice(val any) ([]any, error) {
 	if val == nil {
 		return nil, nil
 	}
-
-	// Try different type assertions for cluster vs regular client compatibility
-	var vals []any
-	if v, ok := val.([]any); ok {
-		vals = v
-	} else if v, ok := val.([]interface{}); ok {
-		vals = make([]any, len(v))
-		for i, item := range v {
-			vals[i] = item
-		}
-	} else {
+	v, ok := val.([]any)
+	if !ok {
 		return nil, fmt.Errorf("mismatched message response type: got %T, value: %v", val, val)
 	}
-	if len(vals) == 0 {
-		return nil, nil
+	return v, nil
+}
+
+// buildQueueMessage converts a {id, message, rc, fr} reply into a QueueMessage.
+func buildQueueMessage(vals []any) (*QueueMessage, error) {
+	if len(vals) < 4 {
+		return nil, fmt.Errorf("malformed message response: %v", vals)
 	}
 	id := vals[0].(string)
 	message := vals[1].(string)