diff --git a/internal/config/config.go b/internal/config/config.go
index 25c4357c..146aea20 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. 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"`
diff --git a/internal/config/config_retry_test.go b/internal/config/config_retry_test.go
index 56bb2dc9..86fb57b4 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,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)
+		})
+	}
+}
diff --git a/internal/config/validation.go b/internal/config/validation.go
index 8347b872..716e93ab 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/rsmq/rsmq.go b/internal/rsmq/rsmq.go
index 7eb10498..1b65a0f5 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 d264a0fc..d5f9e952 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 536975ce..39674ab5 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 86f6a4b9..35f21be8 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,13 @@ 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 — 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
@@ -86,7 +101,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 +115,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 +187,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 +206,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 +235,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 6920f4b5..813d5dbd 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"
@@ -16,7 +17,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 +29,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 +64,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
@@ -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()
@@ -89,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()
 
@@ -96,17 +130,17 @@ 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
 	}
 
 	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{
@@ -120,16 +154,63 @@ 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
+// 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))
+	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 func() {
+		defer close(monitorDone)
+		s.Monitor(ctx)
+	}()
+
+	select {
+	case at := <-done:
+		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")
+	}
 }
 
 func TestScheduler_ParallelMonitor(t *testing.T) {
@@ -139,20 +220,20 @@ 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
 	}
 
 	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{
@@ -166,16 +247,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) {
@@ -185,9 +269,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")
 	}
 
@@ -195,18 +279,18 @@ 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)
 
 	<-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) {
@@ -215,11 +299,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
 		}
 
@@ -228,14 +312,15 @@ 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()
 		})
 
-		return s, &msgs
+		return s, msgs
 	}
 
 	t.Run("different IDs execute independently", func(t *testing.T) {
@@ -250,9 +335,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) {
@@ -270,12 +356,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) {
@@ -288,9 +375,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) {
@@ -305,18 +393,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])
 	})
 }
 
@@ -326,11 +414,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
 		}
 
@@ -339,14 +427,15 @@ 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()
 		})
 
-		return s, &msgs
+		return s, msgs
 	}
 
 	t.Run("cancel removes scheduled task", func(t *testing.T) {
@@ -363,7 +452,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) {
@@ -402,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))
 
@@ -443,9 +531,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
 	}
 
@@ -456,17 +544,16 @@ 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()
 	require.NoError(t, s.Schedule(ctx, id, 0))
 
-	time.Sleep(time.Second)
-	require.Len(t, msgs, 1)
-	require.Equal(t, id, msgs[0])
+	// 3 errors back off 100ms + 200ms + 400ms before the message is received.
+	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) {
diff --git a/internal/services/builder.go b/internal/services/builder.go
index 950f4650..2a101c7c 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)))