From aba4c652e409d36f2786bca552071c3193154c5e Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Fri, 7 Aug 2026 13:13:51 +0700
Subject: [PATCH 1/5] perf(retrymq): cut idle Redis polling to one command per
 interval
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The retry monitor polled every 100ms, and each poll cost 5 client-observable
commands over 2 round trips: a MULTI/HMGET/TIME/EXEC to fetch the queue's
vt/delay/maxsize and the server clock, then the EVALSHA that used them. The
cost was per monitor instance and independent of traffic, so an empty queue
cost as much as a busy one and total load scaled with replica count.

Make receiveMessage self-contained — it reads vt from the :Q hash and calls
TIME itself, which upstream RSMQ could not do under pre-Redis-5 verbatim
script replication. When nothing is due it also returns the time until the
zset's earliest score, so the monitor sleeps until the next message comes due
instead of a flat interval. The zset holds both not-yet-due retries and
in-flight messages hidden by vt, so that score is the correct wake time in
either case, and every sleep is still capped.

RETRY_POLL_BACKOFF_MS is redefined as that cap, default 100 -> 30000, and the
effective value is clamped to the shortest configured retry delay so the idle
interval can never make a retry late. The consecutive-error backoff ladder no
longer derives from it, keeping the ~1 minute of transient-infra tolerance
fixed. At a 30s cap: 129.6M -> 86.4K commands per month, per monitor.

Closes #1014

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 internal/config/config.go            |  21 ++++-
 internal/config/config_retry_test.go |  54 +++++++++++
 internal/rsmq/rsmq.go                | 128 ++++++++++++++++++++-------
 internal/rsmq/rsmq_test.go           |  74 ++++++++++++++++
 internal/rsmq/scripts.go             |  52 ++++++++---
 internal/scheduler/scheduler.go      |  45 ++++++++--
 internal/scheduler/scheduler_test.go |  56 ++++++++++--
 internal/services/builder.go         |   2 +-
 8 files changed, 375 insertions(+), 57 deletions(-)

diff --git a/internal/config/config.go b/internal/config/config.go
index 25c4357c1..80e3888eb 100644
--- a/internal/config/config.go
+++ b/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, so idle cost is one Redis command per interval and worst-case retry lateness is one interval. The effective value is capped at the shortest configured retry delay, so retries are never late. When a retry message is found, the monitor immediately polls for the next message without delay. Default: 30000" 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 = 30000
 	c.RetryVisibilityTimeoutSeconds = 30
 	c.MaxDestinationsPerTenant = 20
 	c.DeliveryTimeoutSeconds = 5
@@ -603,6 +604,22 @@ func (c *Config) GetRetryBackoff() (backoff.Backoff, int) {
 	}, c.RetryMaxLimit
 }
 
+// GetRetryPollBackoff returns the maximum time the retry monitor waits between
+// polls while idle. The configured value is capped at the shortest delay a
+// retry can be scheduled with, 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 {
+	backoff := 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 < backoff {
+		return shortest
+	}
+	return backoff
+}
+
 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"`
diff --git a/internal/config/config_retry_test.go b/internal/config/config_retry_test.go
index 56bb2dc9d..88d978663 100644
--- a/internal/config/config_retry_test.go
+++ b/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,56 @@ retry_interval_seconds: 60
 		})
 	}
 }
+
+func TestGetRetryPollBackoff(t *testing.T) {
+	tests := []struct {
+		name string
+		yaml string
+		want time.Duration
+	}{
+		{
+			name: "defaults to 30s, the default retry interval",
+			yaml: "",
+			want: 30 * time.Second,
+		},
+		{
+			name: "capped at a retry interval shorter than the backoff",
+			yaml: "retry_interval_seconds: 5\n",
+			want: 5 * time.Second,
+		},
+		{
+			name: "capped at the shortest entry of a custom schedule",
+			yaml: "retry_schedule: [10, 5, 300]\n",
+			want: 5 * time.Second,
+		},
+		{
+			name: "a schedule longer than the backoff leaves it alone",
+			yaml: "retry_schedule: [60, 300]\n",
+			want: 30 * time.Second,
+		},
+		{
+			name: "an explicitly configured backoff below the cap is used as-is",
+			yaml: "retry_poll_backoff_ms: 100\n",
+			want: 100 * time.Millisecond,
+		},
+	}
+
+	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())
+		})
+	}
+}
diff --git a/internal/rsmq/rsmq.go b/internal/rsmq/rsmq.go
index 7eb104984..1b65a0f53 100644
--- a/internal/rsmq/rsmq.go
+++ b/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)
diff --git a/internal/rsmq/rsmq_test.go b/internal/rsmq/rsmq_test.go
index d264a0fc2..d5f9e9526 100644
--- a/internal/rsmq/rsmq_test.go
+++ b/internal/rsmq/rsmq_test.go
@@ -474,6 +474,80 @@ func (s *RSMQSuite) TestReceiveMessage() {
 	})
 }
 
+func (s *RSMQSuite) TestReceiveMessagePoll() {
+	t := s.T()
+	qname := "que"
+
+	err := s.rsmq.CreateQueue(qname, UnsetVt, UnsetDelay, UnsetMaxsize)
+	assert.Nil(t, err, "error is not nil on creating a queue")
+
+	t.Run("empty queue reports no next message", func(t *testing.T) {
+		res, err := s.rsmq.ReceiveMessagePoll(qname, UnsetVt)
+		assert.Nil(t, err, "error is not nil on polling an empty queue")
+		assert.Nil(t, res.Message, "message is not nil on polling an empty queue")
+		assert.False(t, res.HasNext, "HasNext is true on polling an empty queue")
+	})
+
+	t.Run("delayed message reports when it comes due", func(t *testing.T) {
+		id, err := s.rsmq.SendMessage(qname, "delayed", 30)
+		assert.Nil(t, err, "error is not nil on sending a delayed message")
+
+		res, err := s.rsmq.ReceiveMessagePoll(qname, UnsetVt)
+		assert.Nil(t, err, "error is not nil on polling for a not-yet-due message")
+		assert.Nil(t, res.Message, "message is not nil while the message is not due")
+		assert.True(t, res.HasNext, "HasNext is false while a message is queued")
+		assert.Greater(t, res.NextDue, 25*time.Second, "NextDue is too small")
+		assert.LessOrEqual(t, res.NextDue, 30*time.Second, "NextDue is too large")
+
+		assert.Nil(t, s.rsmq.DeleteMessage(qname, id), "error is not nil on cleaning up")
+	})
+
+	t.Run("due message is returned", func(t *testing.T) {
+		id, err := s.rsmq.SendMessage(qname, "message", UnsetDelay)
+		assert.Nil(t, err, "error is not nil on sending a message")
+
+		res, err := s.rsmq.ReceiveMessagePoll(qname, UnsetVt)
+		assert.Nil(t, err, "error is not nil on polling for a due message")
+		assert.NotNil(t, res.Message, "message is nil on polling for a due message")
+		assert.Equal(t, id, res.Message.ID, "message ID is not as expected")
+		assert.Equal(t, "message", res.Message.Message, "message body is not as expected")
+		assert.Equal(t, uint64(1), res.Message.Rc, "receive count is not as expected")
+
+		// The message is now hidden by vt rather than removed, so the queue
+		// still reports it as the next one due.
+		res, err = s.rsmq.ReceiveMessagePoll(qname, UnsetVt)
+		assert.Nil(t, err, "error is not nil on polling for a hidden message")
+		assert.Nil(t, res.Message, "message is not nil while hidden by vt")
+		assert.True(t, res.HasNext, "HasNext is false while a message is hidden by vt")
+		assert.Greater(t, res.NextDue, time.Duration(0), "NextDue is not positive while hidden by vt")
+		assert.LessOrEqual(t, res.NextDue, time.Duration(DefaultVt)*time.Second, "NextDue exceeds vt")
+
+		assert.Nil(t, s.rsmq.DeleteMessage(qname, id), "error is not nil on cleaning up")
+	})
+
+	t.Run("explicit vt overrides the queue vt", func(t *testing.T) {
+		id, err := s.rsmq.SendMessage(qname, "message", UnsetDelay)
+		assert.Nil(t, err, "error is not nil on sending a message")
+
+		res, err := s.rsmq.ReceiveMessagePoll(qname, 100)
+		assert.Nil(t, err, "error is not nil on polling with an explicit vt")
+		assert.NotNil(t, res.Message, "message is nil on polling with an explicit vt")
+
+		res, err = s.rsmq.ReceiveMessagePoll(qname, UnsetVt)
+		assert.Nil(t, err, "error is not nil on polling for a hidden message")
+		assert.Nil(t, res.Message, "message is not nil while hidden by the explicit vt")
+		assert.Greater(t, res.NextDue, time.Duration(DefaultVt)*time.Second, "explicit vt was not applied")
+
+		assert.Nil(t, s.rsmq.DeleteMessage(qname, id), "error is not nil on cleaning up")
+	})
+
+	t.Run("error when the queue does not exist", func(t *testing.T) {
+		res, err := s.rsmq.ReceiveMessagePoll("non-existing", UnsetVt)
+		assert.Equal(t, ErrQueueNotFound, err, "error is not as expected")
+		assert.Nil(t, res.Message, "message is not nil on polling a non-existing queue")
+	})
+}
+
 func (s *RSMQSuite) TestPopMessage() {
 	t := s.T()
 	qname := "que"
diff --git a/internal/rsmq/scripts.go b/internal/rsmq/scripts.go
index 536975ce9..39674ab53 100644
--- a/internal/rsmq/scripts.go
+++ b/internal/rsmq/scripts.go
@@ -37,34 +37,66 @@ return o`
 
 // The receiveMessage LUA Script
 //
+// Self-contained: reads the queue's vt from the :Q hash and reads the clock
+// with TIME inside the script, so a receive is a single round trip. (Upstream
+// RSMQ fetched both with a preceding MULTI/HMGET/TIME/EXEC because Redis < 3.2
+// replicated scripts verbatim, which made calling TIME inside a script unsafe.
+// Effect replication has been the default since Redis 5.)
+//
 // Parameters:
 //
 // KEYS[1]: the zset key
 // KEYS[2]: the hash key (zset key + ":Q")
-// ARGV[1]: the current time in ms
-// ARGV[2]: the new calculated time when the vt runs out
+// ARGV[1]: the vt in seconds, or "" to use the queue's configured vt
 //
+// * Resolve vt and the current time
 // * Find a message id
 // * Get the message
 // * Increase the rc (receive count)
 // * Use hset to set the fr (first receive) time
 // * Return the message and the counters
 //
-// Returns:
+// Returns one of:
 //
-// {id, message, rc, fr}
-const scriptReceiveMessage = `local msg = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1], "LIMIT", "0", "1")
+// {"m", id, message, rc, fr}  a message was received
+// {"e", hasNext, msUntilNext} nothing is due; hasNext is 0 for an empty
+// .                           queue, otherwise msUntilNext is the time until
+// .                           the earliest score in the zset
+// {"n"}                       the queue does not exist
+//
+// The zset holds both not-yet-due messages and in-flight messages hidden by vt
+// (a receive re-ZADDs at now+vt rather than removing), so the earliest score is
+// the next moment this queue can yield a message in either case.
+const scriptReceiveMessage = `local vt = redis.call("HGET", KEYS[2], "vt")
+if not vt then
+	return {"n"}
+end
+if ARGV[1] ~= "" then
+	vt = ARGV[1]
+end
+local t = redis.call("TIME")
+local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
+local nowStr = string.format("%d", now)
+local msg = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", nowStr, "LIMIT", "0", "1")
 if #msg == 0 then
-	return {}
+	local nxt = redis.call("ZRANGE", KEYS[1], 0, 0, "WITHSCORES")
+	if #nxt == 0 then
+		return {"e", 0, 0}
+	end
+	local d = math.floor(tonumber(nxt[2]) - now)
+	if d < 0 then
+		d = 0
+	end
+	return {"e", 1, d}
 end
-redis.call("ZADD", KEYS[1], ARGV[2], msg[1])
+redis.call("ZADD", KEYS[1], string.format("%d", now + tonumber(vt) * 1000), msg[1])
 redis.call("HINCRBY", KEYS[2], "totalrecv", 1)
 local mbody = redis.call("HGET", KEYS[2], msg[1])
 local rc = redis.call("HINCRBY", KEYS[2], msg[1] .. ":rc", 1)
-local o = {msg[1], mbody, rc}
+local o = {"m", msg[1], mbody, rc}
 if rc==1 then
-	redis.call("HSET", KEYS[2], msg[1] .. ":fr", ARGV[1])
-	table.insert(o, ARGV[1])
+	redis.call("HSET", KEYS[2], msg[1] .. ":fr", nowStr)
+	table.insert(o, nowStr)
 else
 	local fr = redis.call("HGET", KEYS[2], msg[1] .. ":fr")
 	table.insert(o, fr)
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 86f6a4b90..da64571ea 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -37,6 +37,14 @@ type schedulerImpl struct {
 	exec       func(context.Context, string) error
 }
 
+const (
+	// errorBackoffBase is the first delay of the consecutive-error ladder, and
+	// minIdleSleep the floor on an idle sleep. Both are internal so the error
+	// tolerance documented in New stays fixed regardless of pollBackoff.
+	errorBackoffBase = 100 * time.Millisecond
+	minIdleSleep     = 10 * time.Millisecond
+)
+
 type config struct {
 	visibilityTimeout    uint
 	pollBackoff          time.Duration
@@ -55,6 +63,12 @@ func WithVisibilityTimeout(vt uint) Option {
 	}
 }
 
+// WithPollBackoff sets the maximum time the monitor sleeps when no message is
+// due. When the queue holds a message that is not yet visible, the monitor
+// instead sleeps until that message comes due, so the actual sleep is
+// min(timeUntilNextMessage, backoff). Worst-case lateness is therefore bounded
+// by backoff, and is zero when backoff is at most the shortest delay anything
+// is scheduled with.
 func WithPollBackoff(backoff time.Duration) Option {
 	return func(c *config) {
 		c.pollBackoff = backoff
@@ -86,7 +100,7 @@ func WithLogger(logger *logging.Logger) Option {
 }
 
 func New(name string, rsmqClient rsmq.Client, exec func(context.Context, string) error, opts ...Option) Scheduler {
-	// Error retry schedule (with 100ms pollBackoff used by retrymq):
+	// Error retry schedule:
 	//
 	//   Error  Backoff    Cumulative
 	//   1      100ms      0.1s
@@ -100,7 +114,9 @@ func New(name string, rsmqClient rsmq.Client, exec func(context.Context, string)
 	//   9      15s (cap)  40.5s
 	//   10     15s (cap)  55.5s      ← worker dies (~1 min total)
 	//
-	// Backoff formula: pollBackoff * 2^(attempt-1), capped at maxErrorBackoff.
+	// Backoff formula: errorBackoffBase * 2^(attempt-1), capped at
+	// maxErrorBackoff. The ladder is independent of pollBackoff so that this
+	// tolerance holds however the idle poll interval is configured.
 	// After maxConsecutiveErrors the worker dies permanently (supervisor does
 	// not restart it), so these values must tolerate transient infra outages
 	// (e.g. managed Redis/Dragonfly restarts) without killing the worker.
@@ -170,13 +186,13 @@ func (s *schedulerImpl) Monitor(ctx context.Context) error {
 		case <-ctx.Done():
 			return nil
 		default:
-			msg, err := s.rsmqClient.ReceiveMessage(s.name, rsmq.UnsetVt)
+			res, err := s.rsmqClient.ReceiveMessagePoll(s.name, rsmq.UnsetVt)
 			if err != nil {
 				consecutiveErrors++
 				if consecutiveErrors >= s.config.maxConsecutiveErrors {
 					return fmt.Errorf("max consecutive errors reached: %w", err)
 				}
-				backoff := min(s.config.pollBackoff*time.Duration(1<<(consecutiveErrors-1)), s.config.maxErrorBackoff)
+				backoff := min(errorBackoffBase*time.Duration(1<<(consecutiveErrors-1)), s.config.maxErrorBackoff)
 				s.config.logger.Ctx(ctx).Warn("scheduler receive error, retrying",
 					zap.Error(err),
 					zap.Int("attempt", consecutiveErrors),
@@ -189,8 +205,13 @@ func (s *schedulerImpl) Monitor(ctx context.Context) error {
 				continue
 			}
 			consecutiveErrors = 0
+			msg := res.Message
 			if msg == nil {
-				time.Sleep(s.config.pollBackoff)
+				select {
+				case <-ctx.Done():
+					return nil
+				case <-time.After(s.idleSleep(res)):
+				}
 				continue
 			}
 			if s.config.maxReceiveCount > 0 && msg.Rc > s.config.maxReceiveCount {
@@ -213,6 +234,20 @@ func (s *schedulerImpl) Monitor(ctx context.Context) error {
 	}
 }
 
+// idleSleep returns how long to wait before the next poll when nothing was
+// due. An empty queue waits the full pollBackoff; otherwise the wait is capped
+// at the time until the queue's earliest message becomes visible, so a message
+// scheduled before the monitor started sleeping is picked up on time.
+func (s *schedulerImpl) idleSleep(res rsmq.PollResult) time.Duration {
+	sleep := s.config.pollBackoff
+	if res.HasNext && res.NextDue < sleep {
+		sleep = res.NextDue
+	}
+	// Floor the wait so a message that is due in a millisecond does not turn
+	// the monitor into a spin loop; never above the configured maximum.
+	return max(sleep, min(minIdleSleep, s.config.pollBackoff))
+}
+
 // moveToDLQ dead-letters a message that exceeded the max receive count. The
 // DLQ send happens before the delete so a failure between the two steps can
 // only duplicate the message, not lose it.
diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 6920f4b53..1ad4e2acd 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -16,7 +16,7 @@ import (
 )
 
 // mockRSMQ is a test double that wraps a real RSMQ client and injects
-// transient errors into ReceiveMessage for the first failCount calls.
+// transient errors into ReceiveMessagePoll for the first failCount calls.
 type mockRSMQ struct {
 	inner     *rsmq.RedisSMQ
 	calls     atomic.Int64
@@ -28,11 +28,11 @@ func (m *mockRSMQ) CreateQueue(qname string, vt uint, delay uint, maxsize int) e
 	return m.inner.CreateQueue(qname, vt, delay, maxsize)
 }
 
-func (m *mockRSMQ) ReceiveMessage(qname string, vt uint) (*rsmq.QueueMessage, error) {
+func (m *mockRSMQ) ReceiveMessagePoll(qname string, vt uint) (rsmq.PollResult, error) {
 	if m.calls.Add(1) <= m.failCount {
-		return nil, m.failErr
+		return rsmq.PollResult{}, m.failErr
 	}
-	return m.inner.ReceiveMessage(qname, vt)
+	return m.inner.ReceiveMessagePoll(qname, vt)
 }
 
 func (m *mockRSMQ) SendMessage(qname string, message string, delay uint, opts ...rsmq.SendMessageOption) (string, error) {
@@ -63,14 +63,14 @@ func (m *immediateVisibilityRSMQ) ChangeMessageVisibility(qname string, id strin
 	return m.Client.ChangeMessageVisibility(qname, id, 0)
 }
 
-// alwaysFailRSMQ is a test double that always fails ReceiveMessage.
+// alwaysFailRSMQ is a test double that always fails ReceiveMessagePoll.
 type alwaysFailRSMQ struct {
 	err error
 }
 
 func (m *alwaysFailRSMQ) CreateQueue(string, uint, uint, int) error { return nil }
-func (m *alwaysFailRSMQ) ReceiveMessage(string, uint) (*rsmq.QueueMessage, error) {
-	return nil, m.err
+func (m *alwaysFailRSMQ) ReceiveMessagePoll(string, uint) (rsmq.PollResult, error) {
+	return rsmq.PollResult{}, m.err
 }
 func (m *alwaysFailRSMQ) SendMessage(string, string, uint, ...rsmq.SendMessageOption) (string, error) {
 	return "", nil
@@ -132,6 +132,44 @@ func TestScheduler_Basic(t *testing.T) {
 	require.Equal(t, ids[2], msgs[2])
 }
 
+// TestScheduler_IdleSleepWakesOnDueMessage asserts the monitor sleeps until the
+// next message is due rather than for the full poll backoff. The backoff here
+// is far longer than the test, so a monitor that slept it flat would never run
+// the task.
+func TestScheduler_IdleSleepWakesOnDueMessage(t *testing.T) {
+	t.Parallel()
+
+	redisConfig := testutil.CreateTestRedisConfig(t)
+	rsmqClient := createRSMQClient(t, redisConfig)
+	logger := testutil.CreateTestLogger(t)
+
+	done := make(chan time.Time, 1)
+	exec := func(_ context.Context, id string) error {
+		done <- time.Now()
+		return nil
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	s := scheduler.New("scheduler", rsmqClient, exec,
+		scheduler.WithPollBackoff(time.Minute),
+		scheduler.WithLogger(logger))
+	require.NoError(t, s.Init(ctx))
+	defer func() { cancel(); s.Shutdown() }()
+
+	// Schedule before the monitor starts so its first poll finds the message
+	// pending and has to compute the sleep from the message's due time.
+	require.NoError(t, s.Schedule(ctx, idgen.String(), 1*time.Second))
+	start := time.Now()
+	go s.Monitor(ctx)
+
+	select {
+	case at := <-done:
+		require.WithinDuration(t, start.Add(time.Second), at, 500*time.Millisecond)
+	case <-time.After(5 * time.Second):
+		t.Fatal("task was not executed; monitor slept the full poll backoff")
+	}
+}
+
 func TestScheduler_ParallelMonitor(t *testing.T) {
 	t.Parallel()
 
@@ -464,8 +502,8 @@ func TestScheduler_MonitorRetriesTransientErrors(t *testing.T) {
 	id := idgen.String()
 	require.NoError(t, s.Schedule(ctx, id, 0))
 
-	time.Sleep(time.Second)
-	require.Len(t, msgs, 1)
+	// 3 errors back off 100ms + 200ms + 400ms before the message is received.
+	require.Eventually(t, func() bool { return len(msgs) == 1 }, 5*time.Second, 20*time.Millisecond)
 	require.Equal(t, id, msgs[0])
 }
 
diff --git a/internal/services/builder.go b/internal/services/builder.go
index 950f4650c..2a101c7cc 100644
--- a/internal/services/builder.go
+++ b/internal/services/builder.go
@@ -606,7 +606,7 @@ func (s *serviceInstance) initRetryScheduler(ctx context.Context, cfg *config.Co
 		return fmt.Errorf("log store must be initialized before retry scheduler")
 	}
 	logger.Debug("creating delivery MQ retry scheduler", zap.String("service", s.name))
-	pollBackoff := time.Duration(cfg.RetryPollBackoffMs) * time.Millisecond
+	pollBackoff := cfg.GetRetryPollBackoff()
 	var retrySchedulerOpts []deliverymq.RetrySchedulerOption
 	if cfg.RetryVisibilityTimeoutSeconds > 0 {
 		retrySchedulerOpts = append(retrySchedulerOpts, deliverymq.WithRetryVisibilityTimeout(uint(cfg.RetryVisibilityTimeoutSeconds)))

From 6d5bf5a627c2134ef0c3e500b774c52bc1299c43 Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Mon, 10 Aug 2026 20:11:00 +0700
Subject: [PATCH 2/5] test(scheduler): fix goroutine leak, data race, and tight
 timing in new tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- IdleSleepWakesOnDueMessage: wait for the Monitor goroutine to exit
  after cancel so it cannot log via t after the test completes, and
  assert the execution window as an elapsed range with a looser upper
  bound (3s) for loaded CI.
- MonitorRetriesTransientErrors: guard the msgs slice with a lock
  (msgLog helper) — exec runs on the monitor's goroutines, so the
  require.Eventually read raced with appends under -race.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/scheduler/scheduler_test.go | 41 +++++++++++++++++++++++-----
 1 file changed, 34 insertions(+), 7 deletions(-)

diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 1ad4e2acd..e345a85b2 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -3,6 +3,7 @@ package scheduler_test
 import (
 	"context"
 	"errors"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -79,6 +80,26 @@ func (m *alwaysFailRSMQ) ChangeMessageVisibility(string, string, uint) error { r
 func (m *alwaysFailRSMQ) DeleteMessage(string, string) error                 { return nil }
 func (m *alwaysFailRSMQ) Quit() error                                        { return nil }
 
+// msgLog records messages appended by executor callbacks. The scheduler runs
+// exec on the monitor's goroutines, so test-side reads race with appends
+// without a lock.
+type msgLog struct {
+	mu   sync.Mutex
+	msgs []string
+}
+
+func (l *msgLog) append(msg string) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.msgs = append(l.msgs, msg)
+}
+
+func (l *msgLog) snapshot() []string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return append([]string(nil), l.msgs...)
+}
+
 // createRSMQClient creates an RSMQ client for testing
 func createRSMQClient(t *testing.T, redisConfig *iredis.RedisConfig) *rsmq.RedisSMQ {
 	ctx := context.Background()
@@ -154,17 +175,23 @@ func TestScheduler_IdleSleepWakesOnDueMessage(t *testing.T) {
 		scheduler.WithPollBackoff(time.Minute),
 		scheduler.WithLogger(logger))
 	require.NoError(t, s.Init(ctx))
-	defer func() { cancel(); s.Shutdown() }()
+	monitorDone := make(chan struct{})
+	defer func() { cancel(); <-monitorDone; s.Shutdown() }()
 
 	// Schedule before the monitor starts so its first poll finds the message
 	// pending and has to compute the sleep from the message's due time.
 	require.NoError(t, s.Schedule(ctx, idgen.String(), 1*time.Second))
 	start := time.Now()
-	go s.Monitor(ctx)
+	go func() {
+		defer close(monitorDone)
+		s.Monitor(ctx)
+	}()
 
 	select {
 	case at := <-done:
-		require.WithinDuration(t, start.Add(time.Second), at, 500*time.Millisecond)
+		elapsed := at.Sub(start)
+		require.GreaterOrEqual(t, elapsed, 500*time.Millisecond, "task executed before it was due")
+		require.LessOrEqual(t, elapsed, 3*time.Second, "task executed far past its due time")
 	case <-time.After(5 * time.Second):
 		t.Fatal("task was not executed; monitor slept the full poll backoff")
 	}
@@ -481,9 +508,9 @@ func TestScheduler_MonitorRetriesTransientErrors(t *testing.T) {
 		failErr:   errors.New("connection reset"),
 	}
 
-	msgs := []string{}
+	var msgs msgLog
 	exec := func(_ context.Context, msg string) error {
-		msgs = append(msgs, msg)
+		msgs.append(msg)
 		return nil
 	}
 
@@ -503,8 +530,8 @@ func TestScheduler_MonitorRetriesTransientErrors(t *testing.T) {
 	require.NoError(t, s.Schedule(ctx, id, 0))
 
 	// 3 errors back off 100ms + 200ms + 400ms before the message is received.
-	require.Eventually(t, func() bool { return len(msgs) == 1 }, 5*time.Second, 20*time.Millisecond)
-	require.Equal(t, id, msgs[0])
+	require.Eventually(t, func() bool { return len(msgs.snapshot()) == 1 }, 5*time.Second, 20*time.Millisecond)
+	require.Equal(t, id, msgs.snapshot()[0])
 }
 
 func TestScheduler_MonitorExhaustsRetries(t *testing.T) {

From 455b2e27f20d4ef0fbb60214559d8ef707b1ea37 Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Mon, 10 Aug 2026 20:12:41 +0700
Subject: [PATCH 3/5] test(scheduler): synchronize executor message capture in
 pre-existing tests

The scheduler has always run exec on the monitor's goroutines, so the
unsynchronized msgs slices in TestScheduler_Basic, ParallelMonitor,
VisibilityTimeout, CustomID, and Cancel raced with test-side reads.
These races predate this branch (reproduced on main) but fail the
package under -race. Reuse the msgLog helper everywhere.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/scheduler/scheduler_test.go | 102 +++++++++++++++------------
 1 file changed, 56 insertions(+), 46 deletions(-)

diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index e345a85b2..61c085848 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -117,9 +117,9 @@ func TestScheduler_Basic(t *testing.T) {
 	rsmqClient := createRSMQClient(t, redisConfig)
 	logger := testutil.CreateTestLogger(t)
 
-	msgs := []string{}
+	var msgs msgLog
 	exec := func(_ context.Context, id string) error {
-		msgs = append(msgs, id)
+		msgs.append(id)
 		return nil
 	}
 
@@ -141,16 +141,19 @@ func TestScheduler_Basic(t *testing.T) {
 
 	// Assert
 	time.Sleep(time.Second / 2)
-	require.Len(t, msgs, 0)
+	require.Empty(t, msgs.snapshot())
 	time.Sleep(time.Second)
-	require.Len(t, msgs, 1)
-	require.Equal(t, ids[0], msgs[0])
+	got := msgs.snapshot()
+	require.Len(t, got, 1)
+	require.Equal(t, ids[0], got[0])
 	time.Sleep(time.Second)
-	require.Len(t, msgs, 2)
-	require.Equal(t, ids[1], msgs[1])
+	got = msgs.snapshot()
+	require.Len(t, got, 2)
+	require.Equal(t, ids[1], got[1])
 	time.Sleep(time.Second)
-	require.Len(t, msgs, 3)
-	require.Equal(t, ids[2], msgs[2])
+	got = msgs.snapshot()
+	require.Len(t, got, 3)
+	require.Equal(t, ids[2], got[2])
 }
 
 // TestScheduler_IdleSleepWakesOnDueMessage asserts the monitor sleeps until the
@@ -204,9 +207,9 @@ func TestScheduler_ParallelMonitor(t *testing.T) {
 	rsmqClient := createRSMQClient(t, redisConfig)
 	logger := testutil.CreateTestLogger(t)
 
-	msgs := []string{}
+	var msgs msgLog
 	exec := func(_ context.Context, id string) error {
-		msgs = append(msgs, id)
+		msgs.append(id)
 		return nil
 	}
 
@@ -231,16 +234,19 @@ func TestScheduler_ParallelMonitor(t *testing.T) {
 
 	// Assert
 	time.Sleep(time.Second / 2)
-	require.Len(t, msgs, 0)
+	require.Empty(t, msgs.snapshot())
 	time.Sleep(time.Second)
-	require.Len(t, msgs, 1)
-	require.Equal(t, ids[0], msgs[0])
+	got := msgs.snapshot()
+	require.Len(t, got, 1)
+	require.Equal(t, ids[0], got[0])
 	time.Sleep(time.Second)
-	require.Len(t, msgs, 2)
-	require.Equal(t, ids[1], msgs[1])
+	got = msgs.snapshot()
+	require.Len(t, got, 2)
+	require.Equal(t, ids[1], got[1])
 	time.Sleep(time.Second)
-	require.Len(t, msgs, 3)
-	require.Equal(t, ids[2], msgs[2])
+	got = msgs.snapshot()
+	require.Len(t, got, 3)
+	require.Equal(t, ids[2], got[2])
 }
 
 func TestScheduler_VisibilityTimeout(t *testing.T) {
@@ -250,9 +256,9 @@ func TestScheduler_VisibilityTimeout(t *testing.T) {
 	rsmqClient := createRSMQClient(t, redisConfig)
 	logger := testutil.CreateTestLogger(t)
 
-	msgs := []string{}
+	var msgs msgLog
 	exec := func(_ context.Context, id string) error {
-		msgs = append(msgs, id)
+		msgs.append(id)
 		return errors.New("error")
 	}
 
@@ -268,10 +274,11 @@ func TestScheduler_VisibilityTimeout(t *testing.T) {
 	s.Schedule(ctx, id, 1*time.Second)
 
 	<-ctx.Done()
-	require.Len(t, msgs, 3)
-	require.Equal(t, id, msgs[0])
-	require.Equal(t, id, msgs[1])
-	require.Equal(t, id, msgs[2])
+	got := msgs.snapshot()
+	require.Len(t, got, 3)
+	require.Equal(t, id, got[0])
+	require.Equal(t, id, got[1])
+	require.Equal(t, id, got[2])
 }
 
 func TestScheduler_CustomID(t *testing.T) {
@@ -280,11 +287,11 @@ func TestScheduler_CustomID(t *testing.T) {
 	redisConfig := testutil.CreateTestRedisConfig(t)
 	ctx := context.Background()
 
-	setupTestScheduler := func(t *testing.T) (scheduler.Scheduler, *[]string) {
+	setupTestScheduler := func(t *testing.T) (scheduler.Scheduler, *msgLog) {
 		logger := testutil.CreateTestLogger(t)
-		msgs := []string{}
+		msgs := &msgLog{}
 		exec := func(_ context.Context, task string) error {
-			msgs = append(msgs, task)
+			msgs.append(task)
 			return nil
 		}
 
@@ -300,7 +307,7 @@ func TestScheduler_CustomID(t *testing.T) {
 			s.Shutdown()
 		})
 
-		return s, &msgs
+		return s, msgs
 	}
 
 	t.Run("different IDs execute independently", func(t *testing.T) {
@@ -315,9 +322,10 @@ func TestScheduler_CustomID(t *testing.T) {
 		require.NoError(t, s.Schedule(ctx, task, 0, scheduler.WithTaskID(id2)))
 
 		time.Sleep(time.Second / 2)
-		require.Len(t, *msgs, 2)
-		require.Equal(t, task, (*msgs)[0])
-		require.Equal(t, task, (*msgs)[1])
+		got := msgs.snapshot()
+		require.Len(t, got, 2)
+		require.Equal(t, task, got[0])
+		require.Equal(t, task, got[1])
 	})
 
 	t.Run("same ID overrides previous task and timing", func(t *testing.T) {
@@ -335,12 +343,13 @@ func TestScheduler_CustomID(t *testing.T) {
 
 		// At 1s mark (original task's time), nothing should execute
 		time.Sleep(time.Second + 100*time.Millisecond)
-		require.Empty(t, *msgs, "no task should execute at 1s")
+		require.Empty(t, msgs.snapshot(), "no task should execute at 1s")
 
 		// At 2s mark, only the override should execute
 		time.Sleep(time.Second + 100*time.Millisecond)
-		require.Len(t, *msgs, 1, "override task should execute at 2s")
-		require.Equal(t, task2, (*msgs)[0], "only override task should execute")
+		got := msgs.snapshot()
+		require.Len(t, got, 1, "override task should execute at 2s")
+		require.Equal(t, task2, got[0], "only override task should execute")
 	})
 
 	t.Run("no ID generates unique IDs", func(t *testing.T) {
@@ -353,9 +362,10 @@ func TestScheduler_CustomID(t *testing.T) {
 		require.NoError(t, s.Schedule(ctx, task, 0))
 
 		time.Sleep(time.Second / 2)
-		require.Len(t, *msgs, 2)
-		require.Equal(t, task, (*msgs)[0])
-		require.Equal(t, task, (*msgs)[1])
+		got := msgs.snapshot()
+		require.Len(t, got, 2)
+		require.Equal(t, task, got[0])
+		require.Equal(t, task, got[1])
 	})
 
 	t.Run("ID can be reused after task executes", func(t *testing.T) {
@@ -370,18 +380,18 @@ func TestScheduler_CustomID(t *testing.T) {
 
 		// Wait for first task to execute
 		require.Eventually(t, func() bool {
-			return len(*msgs) >= 1
+			return len(msgs.snapshot()) >= 1
 		}, 2*time.Second, 50*time.Millisecond, "first task should execute")
-		require.Equal(t, task1, (*msgs)[0])
+		require.Equal(t, task1, msgs.snapshot()[0])
 
 		// Schedule second task with same ID
 		require.NoError(t, s.Schedule(ctx, task2, 100*time.Millisecond, scheduler.WithTaskID(id)))
 
 		// Wait for second task to execute
 		require.Eventually(t, func() bool {
-			return len(*msgs) >= 2
+			return len(msgs.snapshot()) >= 2
 		}, 2*time.Second, 50*time.Millisecond, "second task should execute")
-		require.Equal(t, task2, (*msgs)[1])
+		require.Equal(t, task2, msgs.snapshot()[1])
 	})
 }
 
@@ -391,11 +401,11 @@ func TestScheduler_Cancel(t *testing.T) {
 	redisConfig := testutil.CreateTestRedisConfig(t)
 	ctx := context.Background()
 
-	setupTestScheduler := func(t *testing.T) (scheduler.Scheduler, *[]string) {
+	setupTestScheduler := func(t *testing.T) (scheduler.Scheduler, *msgLog) {
 		logger := testutil.CreateTestLogger(t)
-		msgs := []string{}
+		msgs := &msgLog{}
 		exec := func(_ context.Context, task string) error {
-			msgs = append(msgs, task)
+			msgs.append(task)
 			return nil
 		}
 
@@ -411,7 +421,7 @@ func TestScheduler_Cancel(t *testing.T) {
 			s.Shutdown()
 		})
 
-		return s, &msgs
+		return s, msgs
 	}
 
 	t.Run("cancel removes scheduled task", func(t *testing.T) {
@@ -428,7 +438,7 @@ func TestScheduler_Cancel(t *testing.T) {
 
 		// Wait past when it would have executed
 		time.Sleep(time.Second + 100*time.Millisecond)
-		require.Empty(t, *msgs, "cancelled task should not execute")
+		require.Empty(t, msgs.snapshot(), "cancelled task should not execute")
 	})
 
 	t.Run("cancel is idempotent", func(t *testing.T) {

From 147ba9d83d6d36ce3f990779ada908fe92058546 Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Mon, 10 Aug 2026 22:10:52 +0700
Subject: [PATCH 4/5] test(scheduler): wait for monitor goroutines to exit
 before Shutdown
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Test cleanups cancelled the monitor context and called Shutdown without
waiting for the Monitor goroutine to exit. Shutdown breaks the Redis
client mid-poll, so the still-running monitor logs a receive-error Warn
through the zaptest logger after the test has finished — flagged by the
race detector as a write to testing.T past completion.

Add a startMonitor helper that returns a wait function and apply the
cancel → wait → Shutdown ordering at every monitor spawn site.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/scheduler/scheduler_test.go | 46 ++++++++++++++++++----------
 1 file changed, 29 insertions(+), 17 deletions(-)

diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 61c085848..813d5dbd9 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -110,6 +110,19 @@ func createRSMQClient(t *testing.T, redisConfig *iredis.RedisConfig) *rsmq.Redis
 	return rsmq.NewRedisSMQ(adapter, "rsmq")
 }
 
+// startMonitor runs s.Monitor on ctx in a goroutine and returns a wait
+// function that blocks until it has exited. Cancel ctx, then wait, then
+// Shutdown — otherwise the monitor can log through the test logger after the
+// test has finished, which the race detector flags.
+func startMonitor(ctx context.Context, s scheduler.Scheduler) (wait func()) {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		s.Monitor(ctx)
+	}()
+	return func() { <-done }
+}
+
 func TestScheduler_Basic(t *testing.T) {
 	t.Parallel()
 
@@ -126,8 +139,8 @@ func TestScheduler_Basic(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	s := scheduler.New("scheduler", rsmqClient, exec, scheduler.WithLogger(logger))
 	require.NoError(t, s.Init(ctx))
-	defer func() { cancel(); s.Shutdown() }()
-	go s.Monitor(ctx)
+	waitMonitor := startMonitor(ctx, s)
+	defer func() { cancel(); waitMonitor(); s.Shutdown() }()
 
 	// Act
 	ids := []string{
@@ -216,11 +229,11 @@ func TestScheduler_ParallelMonitor(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	s := scheduler.New("scheduler", rsmqClient, exec, scheduler.WithLogger(logger))
 	require.NoError(t, s.Init(ctx))
-	defer func() { cancel(); s.Shutdown() }()
 
-	go s.Monitor(ctx)
-	go s.Monitor(ctx)
-	go s.Monitor(ctx)
+	wait1 := startMonitor(ctx, s)
+	wait2 := startMonitor(ctx, s)
+	wait3 := startMonitor(ctx, s)
+	defer func() { cancel(); wait1(); wait2(); wait3(); s.Shutdown() }()
 
 	// Act
 	ids := []string{
@@ -266,9 +279,8 @@ func TestScheduler_VisibilityTimeout(t *testing.T) {
 	defer cancel()
 	s := scheduler.New("scheduler", rsmqClient, exec, scheduler.WithVisibilityTimeout(1), scheduler.WithLogger(logger))
 	require.NoError(t, s.Init(ctx))
-	defer s.Shutdown()
-
-	go s.Monitor(ctx)
+	waitMonitor := startMonitor(ctx, s)
+	defer func() { cancel(); waitMonitor(); s.Shutdown() }()
 
 	id := idgen.String()
 	s.Schedule(ctx, id, 1*time.Second)
@@ -300,10 +312,11 @@ func TestScheduler_CustomID(t *testing.T) {
 		rsmqClient := createRSMQClient(t, redisConfig)
 		s := scheduler.New(idgen.String(), rsmqClient, exec, scheduler.WithLogger(logger))
 		require.NoError(t, s.Init(ctx))
-		go s.Monitor(monitorCtx)
+		waitMonitor := startMonitor(monitorCtx, s)
 
 		t.Cleanup(func() {
 			cancelMonitor()
+			waitMonitor()
 			s.Shutdown()
 		})
 
@@ -414,10 +427,11 @@ func TestScheduler_Cancel(t *testing.T) {
 		rsmqClient := createRSMQClient(t, redisConfig)
 		s := scheduler.New(idgen.String(), rsmqClient, exec, scheduler.WithLogger(logger))
 		require.NoError(t, s.Init(ctx))
-		go s.Monitor(monitorCtx)
+		waitMonitor := startMonitor(monitorCtx, s)
 
 		t.Cleanup(func() {
 			cancelMonitor()
+			waitMonitor()
 			s.Shutdown()
 		})
 
@@ -477,9 +491,8 @@ func TestScheduler_MaxReceiveCountMovesToDLQ(t *testing.T) {
 		scheduler.WithMaxReceiveCount(2),
 		scheduler.WithLogger(logger))
 	require.NoError(t, s.Init(ctx))
-	defer func() { cancel(); s.Shutdown() }()
-
-	go s.Monitor(ctx)
+	waitMonitor := startMonitor(ctx, s)
+	defer func() { cancel(); waitMonitor(); s.Shutdown() }()
 
 	require.NoError(t, s.Schedule(ctx, task, 0))
 
@@ -531,9 +544,8 @@ func TestScheduler_MonitorRetriesTransientErrors(t *testing.T) {
 		scheduler.WithLogger(logger),
 	)
 	require.NoError(t, s.Init(ctx))
-	defer func() { cancel(); s.Shutdown() }()
-
-	go s.Monitor(ctx)
+	waitMonitor := startMonitor(ctx, s)
+	defer func() { cancel(); waitMonitor(); s.Shutdown() }()
 
 	// Schedule a message — Monitor should recover after 3 transient errors and process it
 	id := idgen.String()

From 5590b770a804d8e422d3699d58e4569f37f5db48 Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Mon, 10 Aug 2026 22:10:52 +0700
Subject: [PATCH 5/5] fix(config): honor explicit retry_poll_backoff_ms; 0
 sentinel = auto
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Capping the configured backoff at the shortest retry delay could
silently override an explicit user value downward (retry_schedule
[5, ...] + explicit 10000ms → forced to 5s, doubling the intended idle
Redis cost). Replace the cap with a sentinel: default 0 means auto —
min(30s, shortest configured retry delay), so retries are never late —
while an explicit positive value is honored as-is as a fixed maximum
idle sleep.

Validation now rejects retry_schedule entries < 1,
retry_interval_seconds < 1 when no schedule is set, and negative
retry_poll_backoff_ms at startup.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/config/config.go            | 24 +++++++----
 internal/config/config_retry_test.go | 62 +++++++++++++++++++++++++---
 internal/config/validation.go        | 11 +++++
 internal/scheduler/scheduler.go      |  3 +-
 4 files changed, 86 insertions(+), 14 deletions(-)

diff --git a/internal/config/config.go b/internal/config/config.go
index 80e3888eb..146aea206 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -81,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:"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, so idle cost is one Redis command per interval and worst-case retry lateness is one interval. The effective value is capped at the shortest configured retry delay, so retries are never late. When a retry message is found, the monitor immediately polls for the next message without delay. Default: 30000" 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
@@ -169,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 = 30000
+	c.RetryPollBackoffMs = 0 // 0 = auto: min(30s, shortest configured retry delay)
 	c.RetryVisibilityTimeoutSeconds = 30
 	c.MaxDestinationsPerTenant = 20
 	c.DeliveryTimeoutSeconds = 5
@@ -604,20 +604,28 @@ 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. The configured value is capped at the shortest delay a
-// retry can be scheduled with, so the monitor is always awake by the time the
-// earliest possible retry comes due and the idle interval never adds latency.
+// 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 {
-	backoff := time.Duration(c.RetryPollBackoffMs) * time.Millisecond
+	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 < backoff {
+	if shortest > 0 && shortest < defaultRetryPollBackoff {
 		return shortest
 	}
-	return backoff
+	return defaultRetryPollBackoff
 }
 
 type TelemetryConfig struct {
diff --git a/internal/config/config_retry_test.go b/internal/config/config_retry_test.go
index 88d978663..86fb57b42 100644
--- a/internal/config/config_retry_test.go
+++ b/internal/config/config_retry_test.go
@@ -114,30 +114,35 @@ func TestGetRetryPollBackoff(t *testing.T) {
 		want time.Duration
 	}{
 		{
-			name: "defaults to 30s, the default retry interval",
+			name: "default is auto, 30s ceiling",
 			yaml: "",
 			want: 30 * time.Second,
 		},
 		{
-			name: "capped at a retry interval shorter than the backoff",
+			name: "auto follows a retry interval shorter than the ceiling",
 			yaml: "retry_interval_seconds: 5\n",
 			want: 5 * time.Second,
 		},
 		{
-			name: "capped at the shortest entry of a custom schedule",
+			name: "auto follows the shortest entry of a custom schedule",
 			yaml: "retry_schedule: [10, 5, 300]\n",
 			want: 5 * time.Second,
 		},
 		{
-			name: "a schedule longer than the backoff leaves it alone",
+			name: "auto stays at the 30s ceiling under a longer schedule",
 			yaml: "retry_schedule: [60, 300]\n",
 			want: 30 * time.Second,
 		},
 		{
-			name: "an explicitly configured backoff below the cap is used as-is",
+			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 {
@@ -159,3 +164,50 @@ func TestGetRetryPollBackoff(t *testing.T) {
 		})
 	}
 }
+
+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)
+		})
+	}
+}
diff --git a/internal/config/validation.go b/internal/config/validation.go
index 8347b8720..716e93ab6 100644
--- a/internal/config/validation.go
+++ b/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)
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index da64571ea..35f21be81 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -68,7 +68,8 @@ func WithVisibilityTimeout(vt uint) Option {
 // instead sleeps until that message comes due, so the actual sleep is
 // min(timeUntilNextMessage, backoff). Worst-case lateness is therefore bounded
 // by backoff, and is zero when backoff is at most the shortest delay anything
-// is scheduled with.
+// is scheduled with — the config auto mode (retry_poll_backoff_ms = 0)
+// guarantees this; explicit values may trade retry latency for idle cost.
 func WithPollBackoff(backoff time.Duration) Option {
 	return func(c *config) {
 		c.pollBackoff = backoff