diff --git a/adapter/sqs.go b/adapter/sqs.go
index 327c79ef1..79d8c1cbe 100644
--- a/adapter/sqs.go
+++ b/adapter/sqs.go
@@ -209,6 +209,11 @@ type SQSServer struct {
// Increment call sites use a nil-receiver-safe call so the
// metrics path costs nothing when unwired.
partitionObserver SQSPartitionObserver
+ // throttleObserver records configured per-queue throttle
+ // outcomes: rejected-request counters and remaining-token gauges.
+ // nil on non-monitored fixtures; observeThrottleDecision is
+ // nil-safe so the request path pays one branch when unwired.
+ throttleObserver SQSThrottleObserver
}
// SQSPartitionObserver is the metrics-package interface
@@ -219,18 +224,48 @@ type SQSPartitionObserver interface {
ObservePartitionMessage(queue string, partition uint32, action string)
}
-// SQSPartitionAction* mirror the action label values from
-// monitoring.SQSPartitionAction*. Re-declared so adapter call
-// sites do not need a monitoring import; the observer interface
-// validates the value at runtime so a drift between these
-// constants and the monitoring side surfaces as a dropped
-// observation rather than a wedge.
+// SQSThrottleObserver is the metrics-package interface
+// (monitoring.SQSThrottleObserver) re-declared here so the adapter
+// does not import monitoring at the package boundary.
+type SQSThrottleObserver interface {
+ ObserveThrottleDecision(queue string, action string, tokensRemaining float64, throttled bool)
+ ForgetThrottleAction(queue string, action string)
+ SyncThrottleActions(queue string, enabledActions []string)
+}
+
+type sqsThrottleRejectionObserver interface {
+ ObserveThrottleRejection(queue string, action string)
+}
+
+type sqsThrottleGaugeCutoffObserver interface {
+ ThrottleGaugeSnapshotCutoff() uint64
+}
+
+type sqsThrottleGaugeResetObserver interface {
+ ForgetThrottleActionBefore(queue string, action string, cutoff uint64)
+}
+
+// SQS metric action labels mirror the values from monitoring/sqs.go.
+// Re-declared so adapter call sites do not need a monitoring import; the
+// observer interface validates the value at runtime so a drift between these
+// constants and the monitoring side surfaces as a dropped observation rather
+// than a wedge.
const (
SQSPartitionActionSend = "send"
SQSPartitionActionReceive = "receive"
SQSPartitionActionDelete = "delete"
+
+ SQSThrottleActionSend = "send"
+ SQSThrottleActionReceive = "receive"
+ SQSThrottleActionDefault = "default"
)
+var sqsThrottleMetricActions = [...]string{
+ SQSThrottleActionSend,
+ SQSThrottleActionReceive,
+ SQSThrottleActionDefault,
+}
+
// WithSQSLeaderMap configures the Raft-address-to-SQS-address mapping used to
// forward requests from followers to the current leader. Format mirrors
// WithDynamoDBLeaderMap / WithS3LeaderMap.
@@ -254,6 +289,14 @@ func WithSQSPartitionObserver(o SQSPartitionObserver) SQSServerOption {
return func(s *SQSServer) { s.partitionObserver = o }
}
+// WithSQSThrottleObserver installs the
+// elastickv_sqs_throttled_requests_total and
+// elastickv_sqs_throttle_tokens_remaining observer on the SQS
+// server. Pass nil (the default) on non-monitored fixtures.
+func WithSQSThrottleObserver(o SQSThrottleObserver) SQSServerOption {
+ return func(s *SQSServer) { s.throttleObserver = o }
+}
+
// observePartitionMessage is a nil-receiver-safe wrapper around
// the configured observer. Pulled into a helper so the call
// sites in send / receive / delete each cost one branch instead
@@ -265,6 +308,100 @@ func (s *SQSServer) observePartitionMessage(queue string, partition uint32, acti
s.partitionObserver.ObservePartitionMessage(queue, partition, action)
}
+func (s *SQSServer) observeThrottleDecision(queue string, outcome chargeOutcome, observeTokens bool) {
+ if s == nil || s.throttleObserver == nil {
+ return
+ }
+ if !outcome.bucketPresent && outcome.allowed {
+ return
+ }
+ action := sqsThrottleMetricAction(outcome.action)
+ if !observeTokens {
+ if !outcome.allowed {
+ if observer, ok := s.throttleObserver.(sqsThrottleRejectionObserver); ok {
+ observer.ObserveThrottleRejection(queue, action)
+ }
+ }
+ return
+ }
+ s.throttleObserver.ObserveThrottleDecision(
+ queue,
+ action,
+ outcome.tokensAfter,
+ !outcome.allowed,
+ )
+}
+
+func (s *SQSServer) observeThrottleConfig(queue string, throttle *sqsQueueThrottle) {
+ if s == nil || s.throttleObserver == nil {
+ return
+ }
+ s.throttleObserver.SyncThrottleActions(queue, enabledThrottleMetricActions(throttle))
+}
+
+func (s *SQSServer) observeThrottleConfigChange(queue string, throttle *sqsQueueThrottle, resetActions []string, resetCutoff uint64) {
+ if s == nil || s.throttleObserver == nil {
+ return
+ }
+ s.throttleObserver.SyncThrottleActions(queue, enabledThrottleMetricActions(throttle))
+ for _, action := range resetActions {
+ if observer, ok := s.throttleObserver.(sqsThrottleGaugeResetObserver); ok {
+ observer.ForgetThrottleActionBefore(queue, action, resetCutoff)
+ continue
+ }
+ s.throttleObserver.ForgetThrottleAction(queue, action)
+ }
+}
+
+func (s *SQSServer) observeThrottleDelete(queue string, resetCutoff uint64) {
+ if s == nil || s.throttleObserver == nil {
+ return
+ }
+ observer, ok := s.throttleObserver.(sqsThrottleGaugeResetObserver)
+ if !ok {
+ s.observeThrottleConfig(queue, nil)
+ return
+ }
+ for _, action := range sqsThrottleMetricActions {
+ observer.ForgetThrottleActionBefore(queue, action, resetCutoff)
+ }
+}
+
+func (s *SQSServer) throttleGaugeSnapshotCutoff() uint64 {
+ if s == nil || s.throttleObserver == nil {
+ return 0
+ }
+ observer, ok := s.throttleObserver.(sqsThrottleGaugeCutoffObserver)
+ if !ok {
+ return 0
+ }
+ return observer.ThrottleGaugeSnapshotCutoff()
+}
+
+func (s *SQSServer) beginThrottleReset(queue string) uint64 {
+ if s == nil || s.throttle == nil {
+ return s.throttleGaugeSnapshotCutoff()
+ }
+ return s.throttle.beginQueueReset(queue, s.throttleGaugeSnapshotCutoff)
+}
+
+func enabledThrottleMetricActions(throttle *sqsQueueThrottle) []string {
+ if throttle == nil || throttle.IsEmpty() {
+ return nil
+ }
+ enabled := make([]string, 0, len(sqsThrottleMetricActions))
+ if throttle.SendCapacity > 0 {
+ enabled = append(enabled, SQSThrottleActionSend)
+ }
+ if throttle.RecvCapacity > 0 {
+ enabled = append(enabled, SQSThrottleActionReceive)
+ }
+ if throttle.DefaultCapacity > 0 {
+ enabled = append(enabled, SQSThrottleActionDefault)
+ }
+ return enabled
+}
+
// WithSQSPartitionResolver installs the cluster's partition
// resolver on the SQS server so the CreateQueue capability gate
// (validateHTFIFOCapability) can verify routing coverage before
diff --git a/adapter/sqs_admin.go b/adapter/sqs_admin.go
index 3b809be64..e3b205dd6 100644
--- a/adapter/sqs_admin.go
+++ b/adapter/sqs_admin.go
@@ -353,7 +353,7 @@ func (s *SQSServer) AdminSetQueueAttributes(ctx context.Context, principal Admin
if strings.TrimSpace(name) == "" || len(attrs) == 0 {
return ErrAdminSQSValidation
}
- throttleChanged, err := s.setQueueAttributesWithRetry(ctx, name, attrs)
+ throttleChanged, throttle, resetActions, throttleResetCutoff, err := s.setQueueAttributesWithRetry(ctx, name, attrs)
if err != nil {
if isSQSAdminQueueDoesNotExist(err) {
return ErrAdminSQSNotFound
@@ -364,7 +364,8 @@ func (s *SQSServer) AdminSetQueueAttributes(ctx context.Context, principal Admin
return errors.Wrap(err, "admin set queue attributes")
}
if throttleChanged {
- s.throttle.invalidateQueue(name)
+ s.throttle.invalidateQueueBuckets(name)
+ s.observeThrottleConfigChange(name, throttle, resetActions, throttleResetCutoff)
}
return nil
}
@@ -383,7 +384,8 @@ func (s *SQSServer) AdminDeleteQueue(ctx context.Context, principal AdminPrincip
if strings.TrimSpace(name) == "" {
return ErrAdminSQSValidation
}
- if err := s.deleteQueueWithRetry(ctx, name); err != nil {
+ throttleResetCutoff, err := s.deleteQueueWithRetry(ctx, name)
+ if err != nil {
// deleteQueueWithRetry returns sqsAPIError with
// sqsErrQueueDoesNotExist when the queue is missing; map
// to the structured ErrAdminSQSNotFound so the admin
@@ -393,6 +395,9 @@ func (s *SQSServer) AdminDeleteQueue(ctx context.Context, principal AdminPrincip
}
return errors.Wrap(err, "admin delete queue")
}
+ s.throttle.invalidateQueueBuckets(name)
+ s.observeThrottleDelete(name, throttleResetCutoff)
+ s.dropReceiveFanoutCounter(name)
return nil
}
diff --git a/adapter/sqs_catalog.go b/adapter/sqs_catalog.go
index 5b5b7f73e..2819e391b 100644
--- a/adapter/sqs_catalog.go
+++ b/adapter/sqs_catalog.go
@@ -538,7 +538,7 @@ var sqsAttributeAppliers = map[string]attributeApplier{
"DeduplicationScope must be 'messageGroup' or 'queue'")
},
// Throttle* are non-AWS extensions for per-queue rate limiting,
- // see docs/design/2026_04_26_proposed_sqs_per_queue_throttling.md.
+ // see docs/design/2026_04_26_implemented_sqs_per_queue_throttling.md.
// Each accepts a non-negative float64; the cross-attribute
// validation that enforces both-zero-or-both-positive on each
// (capacity, refill) pair, capacity ≥ refill, hard ceiling, and
@@ -1043,6 +1043,10 @@ func (s *SQSServer) tryCreateQueueOnce(ctx context.Context, requested *sqsQueueM
{Op: kv.Put, Key: genKey, Value: []byte(strconv.FormatUint(requested.Generation, 10))},
},
}
+ // Open the reset gate before the commit so first requests
+ // against the newly-created incarnation publish gauges newer
+ // than the cleanup cutoff below.
+ throttleResetCutoff := s.beginThrottleReset(requested.Name)
if _, err := s.coordinator.Dispatch(ctx, req); err != nil {
return false, errors.WithStack(err)
}
@@ -1054,7 +1058,8 @@ func (s *SQSServer) tryCreateQueueOnce(ctx context.Context, requested *sqsQueueM
// return path above, which exits before this point) guarantees
// the new queue starts with a fresh full-capacity bucket
// regardless of in-flight traffic to the prior incarnation.
- s.throttle.invalidateQueue(requested.Name)
+ s.throttle.invalidateQueueBuckets(requested.Name)
+ s.observeThrottleConfigChange(requested.Name, requested.Throttle, enabledThrottleMetricActions(requested.Throttle), throttleResetCutoff)
// Mirror the throttle invalidate for the per-queue fanout-rotation
// counter. A delete-then-create race could otherwise leave the
// new queue starting partitioned receives at the previous
@@ -1084,7 +1089,8 @@ func (s *SQSServer) deleteQueue(w http.ResponseWriter, r *http.Request) {
}
func (s *SQSServer) deleteQueueCore(ctx context.Context, name string) error {
- if err := s.deleteQueueWithRetry(ctx, name); err != nil {
+ throttleResetCutoff, err := s.deleteQueueWithRetry(ctx, name)
+ if err != nil {
return err
}
// Drop in-memory throttle buckets belonging to this queue so a
@@ -1094,7 +1100,8 @@ func (s *SQSServer) deleteQueueCore(ctx context.Context, name string) error {
// keep enforcing for up to the idle-evict window (default 1 h),
// surprising operators who use DeleteQueue+CreateQueue to reset
// queue state.
- s.throttle.invalidateQueue(name)
+ s.throttle.invalidateQueueBuckets(name)
+ s.observeThrottleDelete(name, throttleResetCutoff)
// Drop the per-queue fanout-rotation counter as well. Without
// this, repeated DeleteQueue of unique queue names retains one
// receiveFanoutCounters entry per name for the process lifetime
@@ -1103,17 +1110,17 @@ func (s *SQSServer) deleteQueueCore(ctx context.Context, name string) error {
return nil
}
-func (s *SQSServer) deleteQueueWithRetry(ctx context.Context, queueName string) error {
+func (s *SQSServer) deleteQueueWithRetry(ctx context.Context, queueName string) (uint64, error) {
backoff := transactRetryInitialBackoff
deadline := time.Now().Add(transactRetryMaxDuration)
for range transactRetryMaxAttempts {
readTS := s.nextTxnReadTS(ctx)
existing, exists, err := s.loadQueueMetaAt(ctx, queueName, readTS)
if err != nil {
- return errors.WithStack(err)
+ return 0, errors.WithStack(err)
}
if !exists {
- return newSQSAPIError(http.StatusBadRequest, sqsErrQueueDoesNotExist, "queue does not exist")
+ return 0, newSQSAPIError(http.StatusBadRequest, sqsErrQueueDoesNotExist, "queue does not exist")
}
// Bump the generation counter so any stragglers under the old
@@ -1124,7 +1131,7 @@ func (s *SQSServer) deleteQueueWithRetry(ctx context.Context, queueName string)
// keyspace would leak forever.
lastGen, err := s.loadQueueGenerationAt(ctx, queueName, readTS)
if err != nil {
- return errors.WithStack(err)
+ return 0, errors.WithStack(err)
}
metaKey := sqsQueueMetaKey(queueName)
genKey := sqsQueueGenKey(queueName)
@@ -1149,17 +1156,21 @@ func (s *SQSServer) deleteQueueWithRetry(ctx context.Context, queueName string)
{Op: kv.Put, Key: tombstoneKey, Value: tombstoneValue},
},
}
+ // Start the delete cleanup cutoff before the tombstone commit.
+ // A same-name CreateQueue that commits immediately afterward can
+ // then publish token gauges newer than this stale delete cleanup.
+ throttleResetCutoff := s.beginThrottleReset(queueName)
if _, err := s.coordinator.Dispatch(ctx, req); err == nil {
- return nil
+ return throttleResetCutoff, nil
} else if !isRetryableTransactWriteError(err) {
- return errors.WithStack(err)
+ return 0, errors.WithStack(err)
}
if err := waitRetryWithDeadline(ctx, deadline, backoff); err != nil {
- return errors.WithStack(err)
+ return 0, errors.WithStack(err)
}
backoff = nextTransactRetryBackoff(backoff)
}
- return newSQSAPIError(http.StatusInternalServerError, sqsErrInternalFailure, "delete queue retry attempts exhausted")
+ return 0, newSQSAPIError(http.StatusInternalServerError, sqsErrInternalFailure, "delete queue retry attempts exhausted")
}
func (s *SQSServer) listQueues(w http.ResponseWriter, r *http.Request) {
@@ -1532,7 +1543,7 @@ func (s *SQSServer) setQueueAttributesCore(ctx context.Context, name string, att
if len(attrs) == 0 {
return newSQSAPIError(http.StatusBadRequest, sqsErrMissingParameter, "Attributes is required")
}
- throttleChanged, err := s.setQueueAttributesWithRetry(ctx, name, attrs)
+ throttleChanged, throttle, resetActions, throttleResetCutoff, err := s.setQueueAttributesWithRetry(ctx, name, attrs)
if err != nil {
return err
}
@@ -1550,33 +1561,40 @@ func (s *SQSServer) setQueueAttributesCore(ctx context.Context, name string, att
// full capacity. trySetQueueAttributesOnce therefore
// compares the old and new throttle configs under the same Raft
// read snapshot used for the commit and reports whether the values
- // actually moved. The bucket reconciliation in loadOrInit also
+ // actually moved, plus the committed throttle config and post-enabled
+ // actions whose gauges must be reset because the queue-wide invalidation
+ // drops every bucket. This lets the metrics layer clear disabled token
+ // gauges and reset stale enabled-token gauges even if the queue goes
+ // quiet. The bucket reconciliation in loadOrInit also
// catches a stale bucket if a throttle change slips past this gate
// (e.g. via a future admin path), so the gating here is purely a
- // hot-path optimisation plus a no-op-bypass guard.
+ // hot-path optimisation plus a no-op-bypass guard. The reset cutoff
+ // was captured before the successful commit to preserve first
+ // post-commit token observations.
if throttleChanged {
- s.throttle.invalidateQueue(name)
+ s.throttle.invalidateQueueBuckets(name)
+ s.observeThrottleConfigChange(name, throttle, resetActions, throttleResetCutoff)
}
return nil
}
-func (s *SQSServer) setQueueAttributesWithRetry(ctx context.Context, queueName string, attrs map[string]string) (bool, error) {
+func (s *SQSServer) setQueueAttributesWithRetry(ctx context.Context, queueName string, attrs map[string]string) (bool, *sqsQueueThrottle, []string, uint64, error) {
backoff := transactRetryInitialBackoff
deadline := time.Now().Add(transactRetryMaxDuration)
for range transactRetryMaxAttempts {
- throttleChanged, done, err := s.trySetQueueAttributesOnce(ctx, queueName, attrs)
+ throttleChanged, throttle, resetActions, throttleResetCutoff, done, err := s.trySetQueueAttributesOnce(ctx, queueName, attrs)
if err == nil && done {
- return throttleChanged, nil
+ return throttleChanged, throttle, resetActions, throttleResetCutoff, nil
}
if err != nil && !isRetryableTransactWriteError(err) {
- return false, err
+ return false, nil, nil, 0, err
}
if err := waitRetryWithDeadline(ctx, deadline, backoff); err != nil {
- return false, errors.WithStack(err)
+ return false, nil, nil, 0, errors.WithStack(err)
}
backoff = nextTransactRetryBackoff(backoff)
}
- return false, newSQSAPIError(http.StatusInternalServerError, sqsErrInternalFailure, "set queue attributes retry attempts exhausted")
+ return false, nil, nil, 0, newSQSAPIError(http.StatusInternalServerError, sqsErrInternalFailure, "set queue attributes retry attempts exhausted")
}
// applyAndValidateSetAttributes runs the apply + cross-validator
@@ -1614,23 +1632,27 @@ func applyAndValidateSetAttributes(meta *sqsQueueMeta, attrs map[string]string)
return validatePartitionConfig(meta)
}
-// trySetQueueAttributesOnce is one read-validate-commit pass. The
-// returns are (throttleChanged, done, err). done reports whether the
-// caller should stop retrying (the attrs are now committed); an error
-// means either a non-retryable failure (propagate) or a retryable
-// write conflict (retry after backoff). throttleChanged is true iff
-// the post-apply meta's Throttle config differs from the pre-apply
-// snapshot — the caller uses it to gate the cache invalidation so
-// that a no-op same-value SetQueueAttributes does not reset the
-// bucket to full capacity.
-func (s *SQSServer) trySetQueueAttributesOnce(ctx context.Context, queueName string, attrs map[string]string) (bool, bool, error) {
+// trySetQueueAttributesOnce is one read-validate-commit pass. The returns are
+// (throttleChanged, postThrottle, resetActions, resetCutoff, done, err). done reports
+// whether the caller should stop retrying (the attrs are now committed); an
+// error means either a non-retryable failure (propagate) or a retryable write
+// conflict (retry after backoff). throttleChanged is true iff the post-apply
+// meta's Throttle config differs from the pre-apply snapshot — the caller uses
+// it to gate the cache invalidation and metrics reconciliation so that a no-op
+// same-value SetQueueAttributes does not reset the bucket to full capacity or
+// churn token gauges. resetActions is the still-enabled metric-action set
+// whose in-memory bucket was dropped by the queue-wide invalidation and whose
+// stale token gauge must be removed. resetCutoff is captured before the
+// successful Dispatch so post-commit request-path gauges are not removed by
+// the caller's cleanup.
+func (s *SQSServer) trySetQueueAttributesOnce(ctx context.Context, queueName string, attrs map[string]string) (bool, *sqsQueueThrottle, []string, uint64, bool, error) {
readTS := s.nextTxnReadTS(ctx)
meta, exists, err := s.loadQueueMetaAt(ctx, queueName, readTS)
if err != nil {
- return false, false, errors.WithStack(err)
+ return false, nil, nil, 0, false, errors.WithStack(err)
}
if !exists {
- return false, false, newSQSAPIError(http.StatusBadRequest, sqsErrQueueDoesNotExist, "queue does not exist")
+ return false, nil, nil, 0, false, newSQSAPIError(http.StatusBadRequest, sqsErrQueueDoesNotExist, "queue does not exist")
}
// Snapshot the throttle config under the same read TS used for the
// commit so the comparison sees the value the writer is racing
@@ -1638,16 +1660,18 @@ func (s *SQSServer) trySetQueueAttributesOnce(ctx context.Context, queueName str
// floats wrapped in a struct).
preThrottle := snapshotThrottle(meta.Throttle)
if err := applyAndValidateSetAttributes(meta, attrs); err != nil {
- return false, false, err
+ return false, nil, nil, 0, false, err
}
if err := s.validateRedrivePolicyTarget(ctx, meta, readTS); err != nil {
- return false, false, err
+ return false, nil, nil, 0, false, err
}
throttleChanged := !throttleConfigEqual(preThrottle, meta.Throttle)
+ postThrottle := snapshotThrottle(meta.Throttle)
+ resetActions := enabledThrottleMetricActions(postThrottle)
meta.LastModifiedAtMillis = time.Now().UnixMilli()
metaBytes, err := encodeSQSQueueMeta(meta)
if err != nil {
- return false, false, errors.WithStack(err)
+ return false, nil, nil, 0, false, errors.WithStack(err)
}
metaKey := sqsQueueMetaKey(queueName)
// StartTS + ReadKeys prevent two concurrent SetQueueAttributes from
@@ -1661,10 +1685,17 @@ func (s *SQSServer) trySetQueueAttributesOnce(ctx context.Context, queueName str
{Op: kv.Put, Key: metaKey, Value: metaBytes},
},
}
+ var throttleResetCutoff uint64
+ if throttleChanged {
+ // Start the epoch/cutoff gate before Dispatch. A request that
+ // reads the newly-committed throttle config immediately after
+ // this commit must publish a gauge newer than the cleanup cutoff.
+ throttleResetCutoff = s.beginThrottleReset(queueName)
+ }
if _, err := s.coordinator.Dispatch(ctx, req); err != nil {
- return false, false, errors.WithStack(err)
+ return false, nil, nil, 0, false, errors.WithStack(err)
}
- return throttleChanged, true, nil
+ return throttleChanged, postThrottle, resetActions, throttleResetCutoff, true, nil
}
// snapshotThrottle returns a value-copy of the Throttle config so a
diff --git a/adapter/sqs_depth_source.go b/adapter/sqs_depth_source.go
index 018f96476..b405b1627 100644
--- a/adapter/sqs_depth_source.go
+++ b/adapter/sqs_depth_source.go
@@ -22,9 +22,10 @@ type SQSQueueDepth struct {
//
// Returns:
//
-// - (snaps, true) — leader, scrape OK. Observer writes snaps to
-// the gauges and diffs against the previous tick (forgetting
-// any queue that disappeared from this snapshot).
+// - (snaps, true) — leader, scrape OK. snaps is non-nil, even when
+// there are zero queues. Observer writes snaps to the gauges and
+// diffs against the previous tick (forgetting any queue that
+// disappeared from this snapshot).
// - (nil, true) — this node is a follower (leader-only emission
// keeps gauges consistent with AdminListQueues / AdminDescribeQueue
// at the same instant — follower scans would race the leader's
diff --git a/adapter/sqs_messages.go b/adapter/sqs_messages.go
index 51c69c808..516d53d87 100644
--- a/adapter/sqs_messages.go
+++ b/adapter/sqs_messages.go
@@ -474,12 +474,13 @@ func (s *SQSServer) prepareSendMessage(w http.ResponseWriter, r *http.Request) (
// OCC transaction (§4.2): a rejected request never reaches the
// coordinator.
func (s *SQSServer) validateSend(w http.ResponseWriter, r *http.Request, queueName string, in sqsSendMessageInput) (*sqsQueueMeta, uint64, int64, bool) {
+ throttleEpoch := s.throttle.queueEpoch(queueName)
meta, readTS, apiErr := s.loadQueueMetaForSend(r.Context(), queueName, []byte(in.MessageBody))
if apiErr != nil {
writeSQSErrorFromErr(w, apiErr)
return nil, 0, 0, false
}
- if !s.chargeQueueWithThrottle(w, queueName, bucketActionSend, 1, meta.Throttle, meta.Incarnation) {
+ if !s.chargeQueueWithThrottle(w, queueName, bucketActionSend, 1, meta.Throttle, meta.Incarnation, throttleEpoch) {
return nil, 0, 0, false
}
if apiErr := validateMessageAttributes(in.MessageAttributes); apiErr != nil {
@@ -559,11 +560,12 @@ func (s *SQSServer) sendMessage(w http.ResponseWriter, r *http.Request) {
}
func (s *SQSServer) sendMessageCore(ctx context.Context, queueName string, in sqsSendMessageInput) (map[string]string, error) {
+ throttleEpoch := s.throttle.queueEpoch(queueName)
meta, readTS, apiErr := s.loadQueueMetaForSend(ctx, queueName, []byte(in.MessageBody))
if apiErr != nil {
return nil, apiErr
}
- if err := s.chargeQueueWithThrottleErr(queueName, bucketActionSend, 1, meta.Throttle, meta.Incarnation); err != nil {
+ if err := s.chargeQueueWithThrottleErr(queueName, bucketActionSend, 1, meta.Throttle, meta.Incarnation, throttleEpoch); err != nil {
return nil, err
}
if apiErr := validateMessageAttributes(in.MessageAttributes); apiErr != nil {
@@ -809,6 +811,7 @@ func (s *SQSServer) receiveMessage(w http.ResponseWriter, r *http.Request) {
return
}
+ throttleEpoch := s.throttle.queueEpoch(queueName)
readTS := s.nextTxnReadTS(ctx)
meta, exists, err := s.loadQueueMetaAt(ctx, queueName, readTS)
if err != nil {
@@ -823,7 +826,7 @@ func (s *SQSServer) receiveMessage(w http.ResponseWriter, r *http.Request) {
// don't pay an extra meta read just to discover throttling is
// off. Sits AFTER the QueueDoesNotExist branch — a missing queue
// should not consume a Recv token.
- if !s.chargeQueueWithThrottle(w, queueName, bucketActionReceive, 1, meta.Throttle, meta.Incarnation) {
+ if !s.chargeQueueWithThrottle(w, queueName, bucketActionReceive, 1, meta.Throttle, meta.Incarnation, throttleEpoch) {
return
}
max, maxErr := resolveReceiveMaxMessages(in.MaxNumberOfMessages)
@@ -863,6 +866,7 @@ func (s *SQSServer) receiveMessageCore(ctx context.Context, queueName string, in
if _, err := kv.LeaseReadThrough(s.coordinator, ctx); err != nil {
return nil, errors.WithStack(err)
}
+ throttleEpoch := s.throttle.queueEpoch(queueName)
readTS := s.nextTxnReadTS(ctx)
meta, exists, err := s.loadQueueMetaAt(ctx, queueName, readTS)
if err != nil {
@@ -871,7 +875,7 @@ func (s *SQSServer) receiveMessageCore(ctx context.Context, queueName string, in
if !exists {
return nil, newSQSAPIError(http.StatusBadRequest, sqsErrQueueDoesNotExist, "queue does not exist")
}
- if err := s.chargeQueueWithThrottleErr(queueName, bucketActionReceive, 1, meta.Throttle, meta.Incarnation); err != nil {
+ if err := s.chargeQueueWithThrottleErr(queueName, bucketActionReceive, 1, meta.Throttle, meta.Incarnation, throttleEpoch); err != nil {
return nil, err
}
max, maxErr := resolveReceiveMaxMessages(in.MaxNumberOfMessages)
diff --git a/adapter/sqs_throttle.go b/adapter/sqs_throttle.go
index 2cb1fc3b3..353916ff9 100644
--- a/adapter/sqs_throttle.go
+++ b/adapter/sqs_throttle.go
@@ -5,13 +5,14 @@ import (
"math"
"net/http"
"sync"
+ "sync/atomic"
"time"
"github.com/cockroachdb/errors"
)
// Per-queue throttling — token-bucket store that hangs off *SQSServer.
-// See docs/design/2026_04_26_proposed_sqs_per_queue_throttling.md for
+// See docs/design/2026_04_26_implemented_sqs_per_queue_throttling.md for
// the full design and rollout context. This file implements §3.1 (bucket
// store + token bucket), §3.3 (charging model), §3.4 (Throttling
// envelope helpers) and the cache-invalidation primitives §3.1 calls
@@ -122,6 +123,11 @@ type tokenBucket struct {
evicted bool
}
+type bucketQueueEpoch struct {
+ epoch atomic.Uint64
+ updatedUnixNano atomic.Int64
+}
+
// bucketStore holds every active bucket for an SQS server process.
// sync.Map matches the read-mostly access pattern: lookups are nearly
// always Load hits; LoadOrStore pays the write cost only on first use.
@@ -131,6 +137,8 @@ type tokenBucket struct {
// caller of sweep() is the sole goroutine the ticker drives.
type bucketStore struct {
buckets sync.Map // map[bucketKey]*tokenBucket
+ queueEpochMu sync.RWMutex
+ queueEpochs sync.Map // map[string]*bucketQueueEpoch
clock func() time.Time
evictedAfter time.Duration
sweepEvery time.Duration
@@ -159,6 +167,61 @@ func newBucketStoreDefault() *bucketStore {
return newBucketStore(time.Now, throttleIdleEvictAfter)
}
+func (b *bucketStore) queueEpoch(queue string) uint64 {
+ if b == nil || queue == "" {
+ return 0
+ }
+ b.queueEpochMu.RLock()
+ defer b.queueEpochMu.RUnlock()
+ v, ok := b.queueEpochs.Load(queue)
+ if !ok {
+ return 0
+ }
+ epoch, _ := v.(*bucketQueueEpoch)
+ if epoch == nil {
+ return 0
+ }
+ return epoch.epoch.Load()
+}
+
+func (b *bucketStore) bumpQueueEpoch(queue string) {
+ if b == nil || queue == "" {
+ return
+ }
+ b.queueEpochMu.Lock()
+ defer b.queueEpochMu.Unlock()
+ cell := b.queueEpochCellLocked(queue)
+ cell.epoch.Add(1)
+ cell.updatedUnixNano.Store(b.clock().UnixNano())
+}
+
+func (b *bucketStore) beginQueueReset(queue string, cutoff func() uint64) uint64 {
+ if b == nil || queue == "" {
+ if cutoff == nil {
+ return 0
+ }
+ return cutoff()
+ }
+ b.queueEpochMu.Lock()
+ defer b.queueEpochMu.Unlock()
+ var resetCutoff uint64
+ if cutoff != nil {
+ resetCutoff = cutoff()
+ }
+ cell := b.queueEpochCellLocked(queue)
+ cell.epoch.Add(1)
+ cell.updatedUnixNano.Store(b.clock().UnixNano())
+ return resetCutoff
+}
+
+func (b *bucketStore) queueEpochCellLocked(queue string) *bucketQueueEpoch {
+ fresh := &bucketQueueEpoch{}
+ fresh.updatedUnixNano.Store(b.clock().UnixNano())
+ actual, _ := b.queueEpochs.LoadOrStore(queue, fresh)
+ epoch, _ := actual.(*bucketQueueEpoch)
+ return epoch
+}
+
// chargeOutcome is returned from charge so the caller can build the
// Throttling envelope (Retry-After computed from refillRate +
// requestedCount, see §3.4) without re-loading the bucket.
@@ -167,6 +230,7 @@ type chargeOutcome struct {
retryAfter time.Duration
tokensAfter float64
bucketPresent bool
+ action string
}
// charge takes count tokens from the bucket identified by (queue,
@@ -212,6 +276,7 @@ func (b *bucketStore) charge(cfg *sqsQueueThrottle, queue, action string, incarn
bucket := b.loadOrInit(queue, resolvedAction, incarnation, capacity, refill)
outcome, retry := chargeBucket(bucket, b.clock(), count)
if !retry {
+ outcome.action = resolvedAction
return outcome
}
}
@@ -230,6 +295,7 @@ func (b *bucketStore) charge(cfg *sqsQueueThrottle, queue, action string, incarn
allowed: false,
retryAfter: time.Second,
bucketPresent: false,
+ action: resolvedAction,
}
}
@@ -381,6 +447,14 @@ func (b *bucketStore) loadOrInit(queue, action string, incarnation uint64, capac
// fresh full-capacity bucket — a 2x burst on every invalidation
// event.
func (b *bucketStore) invalidateQueue(queue string) {
+ if b == nil {
+ return
+ }
+ b.bumpQueueEpoch(queue)
+ b.invalidateQueueBuckets(queue)
+}
+
+func (b *bucketStore) invalidateQueueBuckets(queue string) {
if b == nil {
return
}
@@ -462,17 +536,46 @@ func (b *bucketStore) runSweepLoop(ctx context.Context) {
// bucket.mu, so there is no AB-BA cycle with charge().
func (b *bucketStore) sweep() {
cutoff := b.clock().Add(-b.evictedAfter)
+ queuesWithBuckets := map[string]struct{}{}
b.buckets.Range(func(k, v any) bool {
+ key, _ := k.(bucketKey)
bucket, _ := v.(*tokenBucket)
bucket.mu.Lock()
if bucket.lastRefill.Before(cutoff) {
if b.buckets.CompareAndDelete(k, v) {
bucket.evicted = true
}
+ } else {
+ queuesWithBuckets[key.queue] = struct{}{}
}
bucket.mu.Unlock()
return true
})
+ b.sweepQueueEpochs(cutoff, queuesWithBuckets)
+}
+
+func (b *bucketStore) sweepQueueEpochs(cutoff time.Time, queuesWithBuckets map[string]struct{}) {
+ if b == nil || b.evictedAfter <= 0 {
+ return
+ }
+ b.queueEpochMu.Lock()
+ defer b.queueEpochMu.Unlock()
+ b.queueEpochs.Range(func(k, v any) bool {
+ queue, _ := k.(string)
+ if _, ok := queuesWithBuckets[queue]; ok {
+ return true
+ }
+ cell, _ := v.(*bucketQueueEpoch)
+ if cell == nil {
+ b.queueEpochs.Delete(k)
+ return true
+ }
+ updated := time.Unix(0, cell.updatedUnixNano.Load())
+ if updated.Before(cutoff) || updated.Equal(cutoff) {
+ b.queueEpochs.Delete(k)
+ }
+ return true
+ })
}
// resolveActionConfig maps a charge() action to (effective bucket
@@ -498,6 +601,19 @@ func resolveActionConfig(cfg *sqsQueueThrottle, action string) (string, float64,
return action, 0, 0
}
+func sqsThrottleMetricAction(action string) string {
+ switch action {
+ case bucketActionSend:
+ return "send"
+ case bucketActionReceive:
+ return "receive"
+ case bucketActionAny:
+ return "default"
+ default:
+ return "default"
+ }
+}
+
// throttleRetryAfterCap bounds the Retry-After value the client sees.
// Without a cap, a tiny refillRate plus
// a large requested count would compute a multi-day wait — and
@@ -589,19 +705,21 @@ func (s *SQSServer) chargeQueue(w http.ResponseWriter, r *http.Request, queueNam
return true
}
-// chargeQueueWithThrottle is the variant for handlers that already
-// have the throttle config in hand from their own meta load. Drops
-// the per-request meta load chargeQueue does, avoiding redundant
-// storage reads on the hot path. incarnation is
-// sqsQueueMeta.Incarnation: it must come from the same meta snapshot
-// the throttle config was read from so a recreate committed
-// mid-request lands in a fresh bucket on the next call rather than
-// mixing tokens with the prior incarnation. NOTE: meta.Incarnation,
-// NOT meta.Generation — PurgeQueue bumps Generation but preserves
-// Incarnation, so keying the bucket by Generation would let a caller
+// chargeQueueWithThrottle is the variant for handlers that already have the
+// throttle config in hand from their own meta load. Drops the per-request meta
+// load chargeQueue does, avoiding redundant storage reads on the hot path.
+// incarnation is sqsQueueMeta.Incarnation: it must come from the same meta
+// snapshot the throttle config was read from so a recreate committed mid-request
+// lands in a fresh bucket on the next call rather than mixing tokens with the
+// prior incarnation. throttleEpoch must be captured before the same meta read;
+// if SetQueueAttributes/Delete/Create invalidates the queue after that snapshot,
+// the request result is still honoured but the stale token-balance metric is not
+// allowed to recreate a gauge the config path already reset. NOTE:
+// meta.Incarnation, NOT meta.Generation — PurgeQueue bumps Generation but
+// preserves Incarnation, so keying the bucket by Generation would let a caller
// bypass the rate limit by repeatedly purging.
-func (s *SQSServer) chargeQueueWithThrottle(w http.ResponseWriter, queueName, action string, count int, throttle *sqsQueueThrottle, incarnation uint64) bool {
- if err := s.chargeQueueWithThrottleErr(queueName, action, count, throttle, incarnation); err != nil {
+func (s *SQSServer) chargeQueueWithThrottle(w http.ResponseWriter, queueName, action string, count int, throttle *sqsQueueThrottle, incarnation uint64, throttleEpoch uint64) bool {
+ if err := s.chargeQueueWithThrottleErr(queueName, action, count, throttle, incarnation, throttleEpoch); err != nil {
writeSQSErrorFromErr(w, err)
return false
}
@@ -612,18 +730,20 @@ func (s *SQSServer) chargeQueueErr(ctx context.Context, queueName, action string
if s.throttle == nil {
return nil
}
+ throttleEpoch := s.throttle.queueEpoch(queueName)
throttle, incarnation, err := s.queueThrottleConfigContext(ctx, queueName)
if err != nil {
return err
}
- return s.chargeQueueWithThrottleErr(queueName, action, count, throttle, incarnation)
+ return s.chargeQueueWithThrottleErr(queueName, action, count, throttle, incarnation, throttleEpoch)
}
-func (s *SQSServer) chargeQueueWithThrottleErr(queueName, action string, count int, throttle *sqsQueueThrottle, incarnation uint64) error {
+func (s *SQSServer) chargeQueueWithThrottleErr(queueName, action string, count int, throttle *sqsQueueThrottle, incarnation uint64, throttleEpoch uint64) error {
if s.throttle == nil {
return nil
}
outcome := s.throttle.charge(throttle, queueName, action, incarnation, count)
+ s.observeThrottleDecision(queueName, outcome, s.throttle.queueEpoch(queueName) == throttleEpoch)
if outcome.allowed {
return nil
}
diff --git a/adapter/sqs_throttle_integration_test.go b/adapter/sqs_throttle_integration_test.go
index 2fd716a27..7649988d7 100644
--- a/adapter/sqs_throttle_integration_test.go
+++ b/adapter/sqs_throttle_integration_test.go
@@ -4,6 +4,8 @@ import (
"net/http"
"strconv"
"testing"
+
+ "github.com/stretchr/testify/require"
)
// === SQS throttle test refill-rate guardrail ============================
@@ -274,6 +276,91 @@ func TestSQSServer_Throttle_SetQueueAttributesInvalidatesBucket(t *testing.T) {
}
}
+func TestSQSServer_Throttle_SetQueueAttributesSyncsDisabledMetrics(t *testing.T) {
+ t.Parallel()
+ nodes, _, _ := createNode(t, 1)
+ defer shutdown(nodes)
+ node := sqsLeaderNode(t, nodes)
+ observer := &recordingSQSThrottleObserver{}
+ node.sqsServer.throttleObserver = observer
+
+ url := mustCreateQueue(t, node, "throttle-metric-sync")
+ mustSetQueueAttributes(t, node, url, map[string]string{
+ "ThrottleSendCapacity": drainBucketCapacity,
+ "ThrottleSendRefillPerSecond": slowRefillRate,
+ "ThrottleRecvCapacity": drainBucketCapacity,
+ "ThrottleRecvRefillPerSecond": slowRefillRate,
+ })
+ require.Contains(t, observer.syncs, throttleSyncReport{
+ queue: "throttle-metric-sync",
+ enabled: []string{SQSThrottleActionSend, SQSThrottleActionReceive},
+ })
+
+ observer.syncs = nil
+ observer.forgotten = nil
+ mustSetQueueAttributes(t, node, url, map[string]string{
+ "ThrottleSendCapacity": "20",
+ "ThrottleSendRefillPerSecond": slowRefillRate,
+ })
+ require.Contains(t, observer.syncs, throttleSyncReport{
+ queue: "throttle-metric-sync",
+ enabled: []string{SQSThrottleActionSend, SQSThrottleActionReceive},
+ })
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "throttle-metric-sync", action: SQSThrottleActionSend})
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "throttle-metric-sync", action: SQSThrottleActionReceive})
+
+ observer.syncs = nil
+ observer.forgotten = nil
+ mustSetQueueAttributes(t, node, url, map[string]string{
+ "ThrottleRecvCapacity": "0",
+ "ThrottleRecvRefillPerSecond": "0",
+ })
+
+ require.Contains(t, observer.syncs, throttleSyncReport{
+ queue: "throttle-metric-sync",
+ enabled: []string{SQSThrottleActionSend},
+ })
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "throttle-metric-sync", action: SQSThrottleActionSend})
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "throttle-metric-sync", action: SQSThrottleActionReceive})
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "throttle-metric-sync", action: SQSThrottleActionDefault})
+}
+
+func TestSQSServer_Throttle_DeleteCreateQueueSyncsMetrics(t *testing.T) {
+ t.Parallel()
+ nodes, _, _ := createNode(t, 1)
+ defer shutdown(nodes)
+ node := sqsLeaderNode(t, nodes)
+ observer := &recordingSQSThrottleObserver{}
+ node.sqsServer.throttleObserver = observer
+
+ queue := "throttle-metric-recreate"
+ url := mustCreateQueue(t, node, queue)
+ mustSetQueueAttributes(t, node, url, newSendThrottleAttrs())
+
+ observer.syncs = nil
+ observer.forgotten = nil
+ if status, _ := callSQS(t, node, sqsDeleteQueueTarget, map[string]any{"QueueUrl": url}); status != http.StatusOK {
+ t.Fatalf("delete: %d", status)
+ }
+ require.Empty(t, observer.syncs,
+ "DeleteQueue cleanup must use cutoff-scoped forgets instead of unconditional action sync")
+ require.ElementsMatch(t, []throttleForgetReport{
+ {queue: queue, action: SQSThrottleActionSend},
+ {queue: queue, action: SQSThrottleActionReceive},
+ {queue: queue, action: SQSThrottleActionDefault},
+ }, observer.forgotten)
+
+ observer.syncs = nil
+ observer.forgotten = nil
+ _ = mustCreateQueue(t, node, queue)
+ require.Contains(t, observer.syncs, throttleSyncReport{queue: queue})
+ require.ElementsMatch(t, []throttleForgetReport{
+ {queue: queue, action: SQSThrottleActionSend},
+ {queue: queue, action: SQSThrottleActionReceive},
+ {queue: queue, action: SQSThrottleActionDefault},
+ }, observer.forgotten)
+}
+
// TestSQSServer_Throttle_DeleteQueueInvalidatesBucket pins the §3.1
// cache-invalidation contract for DeleteQueue: a same-name recreate
// gets a fresh bucket, not the stale balance from the previous
diff --git a/adapter/sqs_throttle_test.go b/adapter/sqs_throttle_test.go
index eddc732d4..314c2432b 100644
--- a/adapter/sqs_throttle_test.go
+++ b/adapter/sqs_throttle_test.go
@@ -1,6 +1,8 @@
package adapter
import (
+ "net/http"
+ "net/http/httptest"
"sync"
"sync/atomic"
"testing"
@@ -9,6 +11,94 @@ import (
"github.com/stretchr/testify/require"
)
+type throttleObserveReport struct {
+ queue string
+ action string
+ tokensRemaining float64
+ throttled bool
+}
+
+type throttleForgetReport struct {
+ queue string
+ action string
+}
+
+type throttleSyncReport struct {
+ queue string
+ enabled []string
+}
+
+type recordingSQSThrottleObserver struct {
+ reports []throttleObserveReport
+ rejections []throttleForgetReport
+ forgotten []throttleForgetReport
+ syncs []throttleSyncReport
+ seq uint64
+ actionSeq map[throttleForgetReport]uint64
+}
+
+func (r *recordingSQSThrottleObserver) ObserveThrottleDecision(queue string, action string, tokensRemaining float64, throttled bool) {
+ r.seq++
+ if r.actionSeq == nil {
+ r.actionSeq = map[throttleForgetReport]uint64{}
+ }
+ r.actionSeq[throttleForgetReport{queue: queue, action: action}] = r.seq
+ r.reports = append(r.reports, throttleObserveReport{
+ queue: queue,
+ action: action,
+ tokensRemaining: tokensRemaining,
+ throttled: throttled,
+ })
+}
+
+func (r *recordingSQSThrottleObserver) ObserveThrottleRejection(queue string, action string) {
+ r.rejections = append(r.rejections, throttleForgetReport{queue: queue, action: action})
+}
+
+func (r *recordingSQSThrottleObserver) ForgetThrottleAction(queue string, action string) {
+ if r.actionSeq != nil {
+ delete(r.actionSeq, throttleForgetReport{queue: queue, action: action})
+ }
+ r.forgotten = append(r.forgotten, throttleForgetReport{queue: queue, action: action})
+}
+
+func (r *recordingSQSThrottleObserver) ForgetThrottleActionBefore(queue string, action string, cutoff uint64) {
+ key := throttleForgetReport{queue: queue, action: action}
+ if r.actionSeq != nil {
+ if seq := r.actionSeq[key]; seq > cutoff {
+ return
+ }
+ delete(r.actionSeq, key)
+ }
+ r.forgotten = append(r.forgotten, key)
+}
+
+func (r *recordingSQSThrottleObserver) ThrottleGaugeSnapshotCutoff() uint64 {
+ return r.seq
+}
+
+func (r *recordingSQSThrottleObserver) SyncThrottleActions(queue string, enabledActions []string) {
+ enabled := append([]string(nil), enabledActions...)
+ r.syncs = append(r.syncs, throttleSyncReport{queue: queue, enabled: enabled})
+ for _, action := range disabledThrottleMetricActionsFromEnabled(enabled) {
+ r.ForgetThrottleAction(queue, action)
+ }
+}
+
+func disabledThrottleMetricActionsFromEnabled(enabledActions []string) []string {
+ enabled := make(map[string]struct{}, len(enabledActions))
+ for _, action := range enabledActions {
+ enabled[action] = struct{}{}
+ }
+ disabled := make([]string, 0, len(sqsThrottleMetricActions))
+ for _, action := range sqsThrottleMetricActions {
+ if _, ok := enabled[action]; !ok {
+ disabled = append(disabled, action)
+ }
+ }
+ return disabled
+}
+
// TestBucketStore_DefaultOff_ShortCircuit pins the contract that a
// nil throttle config never allocates a bucket and never rejects.
// This is the hot path for unconfigured queues — every nil-check that
@@ -24,6 +114,131 @@ func TestBucketStore_DefaultOff_ShortCircuit(t *testing.T) {
}
}
+func TestSQSServer_ChargeQueueWithThrottleObservesMetrics(t *testing.T) {
+ t.Parallel()
+ observer := &recordingSQSThrottleObserver{}
+ srv := NewSQSServer(nil, nil, nil, WithSQSThrottleObserver(observer))
+ now := time.Date(2026, 4, 27, 10, 0, 0, 0, time.UTC)
+ srv.throttle = newBucketStore(func() time.Time { return now }, throttleIdleEvictAfter)
+ cfg := &sqsQueueThrottle{SendCapacity: 10, SendRefillPerSecond: 1}
+
+ for range 10 {
+ rec := httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, 0))
+ }
+ rec := httptest.NewRecorder()
+ require.False(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, 0))
+ require.Equal(t, http.StatusBadRequest, rec.Code)
+
+ require.Len(t, observer.reports, 11)
+ last := observer.reports[len(observer.reports)-1]
+ require.Equal(t, "orders.fifo", last.queue)
+ require.Equal(t, "send", last.action)
+ require.True(t, last.throttled)
+ require.InDelta(t, 0.0, last.tokensRemaining, 0.001)
+}
+
+func TestSQSServer_ChargeQueueWithThrottleDoesNotSyncMetricActions(t *testing.T) {
+ t.Parallel()
+ observer := &recordingSQSThrottleObserver{}
+ srv := NewSQSServer(nil, nil, nil, WithSQSThrottleObserver(observer))
+ cfg := &sqsQueueThrottle{SendCapacity: 10, SendRefillPerSecond: 1}
+
+ rec := httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, 0))
+ require.Empty(t, observer.syncs, "request-path throttle decisions must not sync enabled actions from a potentially stale meta snapshot")
+ require.Empty(t, observer.forgotten, "request-path throttle decisions must not forget gauges from a potentially stale meta snapshot")
+
+ observer.forgotten = nil
+ observer.syncs = nil
+ rec = httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, nil, 1, 0))
+ require.Empty(t, observer.syncs)
+ require.Empty(t, observer.forgotten)
+}
+
+func TestSQSServer_ChargeQueueWithThrottleSkipsStaleMetricAfterInvalidate(t *testing.T) {
+ t.Parallel()
+ observer := &recordingSQSThrottleObserver{}
+ srv := NewSQSServer(nil, nil, nil, WithSQSThrottleObserver(observer))
+ now := time.Date(2026, 4, 27, 10, 0, 0, 0, time.UTC)
+ srv.throttle = newBucketStore(func() time.Time { return now }, throttleIdleEvictAfter)
+ cfg := &sqsQueueThrottle{SendCapacity: 10, SendRefillPerSecond: 1}
+
+ staleEpoch := srv.throttle.queueEpoch("orders.fifo")
+ srv.throttle.invalidateQueue("orders.fifo")
+
+ rec := httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, staleEpoch))
+ require.Empty(t, observer.reports, "stale pre-invalidation snapshot must not recreate a reset token gauge")
+
+ freshEpoch := srv.throttle.queueEpoch("orders.fifo")
+ rec = httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, freshEpoch))
+ require.Len(t, observer.reports, 1, "fresh post-invalidation snapshot may publish the new token gauge")
+ require.Equal(t, "orders.fifo", observer.reports[0].queue)
+ require.Equal(t, SQSThrottleActionSend, observer.reports[0].action)
+}
+
+func TestSQSServer_BeginThrottleResetSuppressesStaleMetricBeforeCleanup(t *testing.T) {
+ t.Parallel()
+ observer := &recordingSQSThrottleObserver{}
+ srv := NewSQSServer(nil, nil, nil, WithSQSThrottleObserver(observer))
+ now := time.Date(2026, 4, 27, 10, 0, 0, 0, time.UTC)
+ srv.throttle = newBucketStore(func() time.Time { return now }, throttleIdleEvictAfter)
+ cfg := &sqsQueueThrottle{SendCapacity: 10, SendRefillPerSecond: 1}
+
+ staleEpoch := srv.throttle.queueEpoch("orders.fifo")
+ resetCutoff := srv.beginThrottleReset("orders.fifo")
+ rec := httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, staleEpoch))
+ require.Empty(t, observer.reports, "pre-reset epoch request must not publish a token gauge after the reset gate starts")
+
+ srv.throttle.invalidateQueueBuckets("orders.fifo")
+ freshEpoch := srv.throttle.queueEpoch("orders.fifo")
+ rec = httptest.NewRecorder()
+ require.True(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 1, cfg, 1, freshEpoch))
+ require.Len(t, observer.reports, 1, "fresh post-reset request may publish a token gauge")
+
+ srv.observeThrottleConfigChange("orders.fifo", cfg, []string{SQSThrottleActionSend}, resetCutoff)
+ require.NotContains(t, observer.forgotten, throttleForgetReport{queue: "orders.fifo", action: SQSThrottleActionSend},
+ "reset cleanup must preserve token gauges observed after the reset gate")
+}
+
+func TestSQSServer_ObserveThrottleDeletePreservesFreshGaugeAfterCutoff(t *testing.T) {
+ t.Parallel()
+ observer := &recordingSQSThrottleObserver{}
+ srv := NewSQSServer(nil, nil, nil, WithSQSThrottleObserver(observer))
+
+ observer.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 1, false)
+ deleteCutoff := observer.ThrottleGaugeSnapshotCutoff()
+ observer.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 9, false)
+
+ srv.observeThrottleDelete("orders.fifo", deleteCutoff)
+
+ require.NotContains(t, observer.forgotten, throttleForgetReport{queue: "orders.fifo", action: SQSThrottleActionSend},
+ "stale delete cleanup must not erase a same-name new incarnation gauge observed after the cutoff")
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "orders.fifo", action: SQSThrottleActionReceive})
+ require.Contains(t, observer.forgotten, throttleForgetReport{queue: "orders.fifo", action: SQSThrottleActionDefault})
+}
+
+func TestSQSServer_ChargeQueueWithThrottleCountsStaleRejectionAfterInvalidate(t *testing.T) {
+ t.Parallel()
+ observer := &recordingSQSThrottleObserver{}
+ srv := NewSQSServer(nil, nil, nil, WithSQSThrottleObserver(observer))
+ now := time.Date(2026, 4, 27, 10, 0, 0, 0, time.UTC)
+ srv.throttle = newBucketStore(func() time.Time { return now }, throttleIdleEvictAfter)
+ cfg := &sqsQueueThrottle{SendCapacity: 1, SendRefillPerSecond: 1}
+
+ staleEpoch := srv.throttle.queueEpoch("orders.fifo")
+ srv.throttle.invalidateQueue("orders.fifo")
+
+ rec := httptest.NewRecorder()
+ require.False(t, srv.chargeQueueWithThrottle(rec, "orders.fifo", bucketActionSend, 2, cfg, 1, staleEpoch))
+ require.Empty(t, observer.reports, "stale pre-invalidation snapshot must not recreate a reset token gauge")
+ require.Equal(t, []throttleForgetReport{{queue: "orders.fifo", action: SQSThrottleActionSend}}, observer.rejections)
+}
+
// TestBucketStore_Empty_ShortCircuit covers the post-validator
// canonicalisation path: an all-zero sqsQueueThrottle is equivalent
// to nil. Without this branch, a queue whose operator wrote
@@ -641,6 +856,22 @@ func TestBucketStore_InvalidateQueueClearsAllIncarnations(t *testing.T) {
require.True(t, hasEvents, "unrelated queue must not be evicted")
}
+func TestBucketStore_SweepEvictsIdleQueueEpochs(t *testing.T) {
+ t.Parallel()
+ now := time.Date(2026, 4, 27, 10, 0, 0, 0, time.UTC)
+ store := newBucketStore(func() time.Time { return now }, time.Minute)
+
+ store.invalidateQueue("orders")
+ require.Equal(t, uint64(1), store.queueEpoch("orders"))
+ require.Equal(t, 1, countQueueEpochs(store))
+
+ now = now.Add(2 * time.Minute)
+ store.sweep()
+
+ require.Equal(t, uint64(0), store.queueEpoch("orders"))
+ require.Equal(t, 0, countQueueEpochs(store), "idle epoch cells must not grow without bound under queue churn")
+}
+
// TestBucketStore_PurgeQueueDoesNotResetBucket pins the
// PurgeQueue-bypass guard: PurgeQueue bumps sqsQueueMeta.Generation
// but preserves Incarnation, and the throttle bucket keys by Incarnation.
@@ -683,6 +914,15 @@ func countBuckets(b *bucketStore) int {
return n
}
+func countQueueEpochs(b *bucketStore) int {
+ n := 0
+ b.queueEpochs.Range(func(_, _ any) bool {
+ n++
+ return true
+ })
+ return n
+}
+
func TestBucketStore_InvalidateQueueDropsAllActions(t *testing.T) {
t.Parallel()
cfg := &sqsQueueThrottle{
diff --git a/docs/design/2026_04_24_partial_sqs_compatible_adapter.md b/docs/design/2026_04_24_partial_sqs_compatible_adapter.md
index 2140d58b2..47d59120d 100644
--- a/docs/design/2026_04_24_partial_sqs_compatible_adapter.md
+++ b/docs/design/2026_04_24_partial_sqs_compatible_adapter.md
@@ -463,10 +463,12 @@ New metrics (prefixed `sqs_`):
5. `sqs_messages_received_total{queue, is_fifo}`
6. `sqs_messages_deleted_total{queue}`
7. `sqs_messages_in_flight{queue}`
-8. `sqs_longpoll_waiters{queue}`
-9. `sqs_longpoll_wake_latency_seconds`
-10. `sqs_dlq_redrive_total{queue}`
-11. `sqs_proxy_to_leader_total{op}`
+8. `sqs_throttled_requests_total{queue, action}`
+9. `sqs_throttle_tokens_remaining{queue, action}`
+10. `sqs_longpoll_waiters{queue}`
+11. `sqs_longpoll_wake_latency_seconds`
+12. `sqs_dlq_redrive_total{queue}`
+13. `sqs_proxy_to_leader_total{op}`
Structured log fields match the rest of the project: `queue`, `message_id`, `receipt_token_prefix`, `group_id`, `dedup_id`, `commit_ts`, `leader`.
@@ -690,7 +692,7 @@ Structured logs include `route`, `queue`, `action`, `remote_ip`, `token_hash_pre
1. Query-protocol XML support for older SDKs is wired for the supported public verbs; see [`2026_04_26_implemented_sqs_query_protocol.md`](2026_04_26_implemented_sqs_query_protocol.md).
2. `ApproximateNumberOfMessagesDelayed` / `NotVisible` accuracy.
-3. Per-queue throttling and fairness across tenants.
+3. Per-queue throttling has landed; see [`2026_04_26_implemented_sqs_per_queue_throttling.md`](2026_04_26_implemented_sqs_per_queue_throttling.md).
4. Split-queue FIFO for very hot queues.
5. Console UI polish: in-flight tab with per-message countdown, filtering, dark mode.
diff --git a/docs/design/2026_04_26_proposed_sqs_per_queue_throttling.md b/docs/design/2026_04_26_implemented_sqs_per_queue_throttling.md
similarity index 69%
rename from docs/design/2026_04_26_proposed_sqs_per_queue_throttling.md
rename to docs/design/2026_04_26_implemented_sqs_per_queue_throttling.md
index 91e66be67..2595236a5 100644
--- a/docs/design/2026_04_26_proposed_sqs_per_queue_throttling.md
+++ b/docs/design/2026_04_26_implemented_sqs_per_queue_throttling.md
@@ -1,6 +1,6 @@
# Per-Queue Throttling and Tenant Fairness for the SQS Adapter
-**Status:** Proposed
+**Status:** Implemented
**Author:** bootjp
**Date:** 2026-04-26
@@ -8,15 +8,15 @@
## 1. Background and Motivation
-elastickv's SQS adapter currently has **no per-queue rate limiting**. A single tenant's runaway producer can:
+Before this work, elastickv's SQS adapter had **no per-queue rate limiting**. A single tenant's runaway producer could:
1. Saturate the leader's Raft proposal pipeline (one OCC dispatch per `SendMessage`), pushing latency on every other queue's writes through the same shard.
2. Exhaust the receive-path's visibility-index scan budget (`sqsVisScanPageLimit = 1024`), causing other tenants' `ReceiveMessage` calls to time out empty.
3. Fill the message keyspace fast enough that the retention reaper cannot keep up — the keyspace grows unbounded until the next manual purge.
-Phase 3.C in [`docs/design/2026_04_24_partial_sqs_compatible_adapter.md`](2026_04_24_partial_sqs_compatible_adapter.md) §16.5 marks this as TODO. AWS itself enforces per-account / per-API limits ("standard request throttle of 3000 RPS per region per AWS account" plus per-API limits like 300 TPS for batch APIs); operators running elastickv as a multi-tenant SQS facade need an equivalent control plane. Without it, the only knobs are (a) shard-level capacity (too coarse — adding a shard requires a Raft membership change) and (b) external load-balancer rate limiting (no visibility into per-queue cost).
+Phase 3.C in [`docs/design/2026_04_24_proposed_sqs_compatible_adapter.md`](2026_04_24_proposed_sqs_compatible_adapter.md) §14 originally listed this as future work. AWS itself enforces per-account / per-API limits ("standard request throttle of 3000 RPS per region per AWS account" plus per-API limits like 300 TPS for batch APIs); operators running elastickv as a multi-tenant SQS facade need an equivalent control plane. Without it, the only knobs are (a) shard-level capacity (too coarse — adding a shard requires a Raft membership change) and (b) external load-balancer rate limiting (no visibility into per-queue cost).
-This document proposes per-queue token-bucket throttling, configured per-queue in queue meta, evaluated at the SQS-adapter layer on the leader, and surfaced as the same `Throttling.Sender` error AWS uses (so existing SDK retry/backoff logic engages naturally).
+This document describes per-queue token-bucket throttling, configured per-queue in queue meta, evaluated at the SQS-adapter layer on the leader, and surfaced as the same `Throttling.Sender` error AWS uses (so existing SDK retry/backoff logic engages naturally).
---
@@ -26,10 +26,10 @@ This document proposes per-queue token-bucket throttling, configured per-queue i
1. **Per-queue rate limits** that an operator can set via `SetQueueAttributes` and read back via `GetQueueAttributes`. Limits are persisted on the queue meta record (one Raft commit, no separate keyspace).
2. **Per-action granularity** — `SendMessage` and `ReceiveMessage` have independent buckets so a slow consumer cannot pin the producer or vice versa. Batch verbs charge by entry count, not by call count.
-3. **AWS-shape errors**: throttled requests return HTTP `400` with the `Throttling` error code in whichever envelope the request protocol uses (`{"__type":"Throttling", ...}` for the JSON path, `Throttling` for the query/XML path — see §3.4 for the exact wire shape per protocol) and a `Retry-After` header. SDKs already special-case this code with exponential backoff; we do not invent a new code.
+3. **AWS-shape errors**: throttled JSON-path requests return HTTP `400` with the `Throttling` error code (`{"__type":"Throttling", ...}`) and a `Retry-After` header. SDKs already special-case this code with exponential backoff; we do not invent a new code. Query/XML message verbs are not claimed by this lifecycle update until the query dispatcher wires those verbs into the same throttled handlers.
4. **Default-off**. Queues created before this feature, and queues created without explicit limits, are not throttled. Operators opt in per queue.
5. **No coordination per request**. Token replenishment is local to whichever node owns the bucket (the leader for the queue's shard); there is no Raft round-trip on the throttling check.
-6. **Observable**: per-queue throttle counters are exposed via the existing Prometheus registry so dashboards can spot throttling before users do.
+6. **Observable throttling outcome**: throttled requests use the AWS-shaped `Throttling` error and `Retry-After` header so clients can observe and back off from the limit immediately. The server also emits `elastickv_sqs_throttled_requests_total{queue, action}` and `elastickv_sqs_throttle_tokens_remaining{queue, action}` so operators can alert on rejects and inspect the current bucket balance.
### 2.2 Non-Goals
@@ -66,14 +66,14 @@ The check sits **between SigV4 authorisation and the existing handler dispatch**
- the queue name (parsed from the request body's `QueueUrl` / `QueueName`);
- this node is the verified leader for the queue's shard (`isVerifiedSQSLeader`); any non-leader has been forwarded by `proxyToLeader` and re-evaluates the limit on landing;
-- the action (`X-Amz-Target` for JSON, `Action` form parameter for query).
+- the action (`X-Amz-Target` for the implemented JSON path; future query/XML message-verb wiring will use the `Action` form parameter).
### 3.1 Where the bucket lives
A `bucketStore` instance hangs off `*SQSServer`. Internally:
```go
-// adapter/sqs_throttle.go (new in implementation PR)
+// adapter/sqs_throttle.go
type bucketStore struct {
// sync.Map rather than a single mu+map so the hot SendMessage /
// ReceiveMessage path does not contend on a process-wide lock.
@@ -83,7 +83,7 @@ type bucketStore struct {
// pair. Each bucket's own mutation (charge / refill) is guarded by
// a per-bucket sync.Mutex inside *tokenBucket, scoped to one queue,
// so cross-queue traffic never serialises on the same lock.
- // (Gemini medium on PR #664 flagged a single-mutex bucket store as
+ // (A medium review finding on PR #664 flagged a single-mutex bucket store as
// a hot-path contention point; this design avoids that.)
buckets sync.Map // map[bucketKey]*tokenBucket
clock func() time.Time
@@ -112,13 +112,13 @@ The `charge` operation:
No global lock is held during step 3; concurrent traffic on different queues runs in parallel.
-**Cache invalidation on `SetQueueAttributes`**: when an operator updates the throttle config via `SetQueueAttributes`, the handler — *after* the Raft commit that persists the new `sqsQueueThrottle` — calls `buckets.invalidateQueue(name)` (the same path described in the `DeleteQueue` / `CreateQueue` paragraph below). `invalidateQueue` ranges the map and drops every entry whose `queue` matches under a lock-then-`CompareAndDelete`-then-`evicted=true` ordering; a raw per-key `buckets.Delete(key)` would reintroduce the orphan-bucket race that ordering closes (a charger holding the old pointer pre-Delete acquires the bucket's mu after the map entry is gone, sees `evicted=false`, spends a token, then later requests mint a fresh full-capacity bucket — a transient double-allotment window). Without this step at all, the in-memory bucket would keep enforcing the old limits until the idle-eviction sweep removes the stale entry (default 1 h window), defeating the operator's intent to throttle a noisy tenant in real time. The handler also gates the invalidation on a real value change — a same-value `SetQueueAttributes` does not reset the bucket — so a caller cannot bypass the rate limit by re-submitting their own current config. Claude P1 on PR #664 caught the gap; round 9 / round 12 refined the race-free semantics and the no-op gate.
+**Cache invalidation on `SetQueueAttributes`**: when an operator updates the throttle config via `SetQueueAttributes`, the handler — *after* the Raft commit that persists the new `sqsQueueThrottle` — calls `buckets.invalidateQueue(name)` (the same path described in the `DeleteQueue` / `CreateQueue` paragraph below). `invalidateQueue` ranges the map and drops every entry whose `queue` matches under a lock-then-`CompareAndDelete`-then-`evicted=true` ordering; a raw per-key `buckets.Delete(key)` would reintroduce the orphan-bucket race that ordering closes (a charger holding the old pointer pre-Delete acquires the bucket's mu after the map entry is gone, sees `evicted=false`, spends a token, then later requests mint a fresh full-capacity bucket — a transient double-allotment window). Without this step at all, the in-memory bucket would keep enforcing the old limits until the idle-eviction sweep removes the stale entry (default 1 h window), defeating the operator's intent to throttle a noisy tenant in real time. The handler also gates the invalidation on a real value change — a same-value `SetQueueAttributes` does not reset the bucket — so a caller cannot bypass the rate limit by re-submitting their own current config. A P1 review finding on PR #664 caught the gap; round 9 / round 12 refined the race-free semantics and the no-op gate.
**Cache invalidation on `DeleteQueue` / `CreateQueue`**: when a queue is deleted, the handler — *after* the Raft commit that purges the queue meta — calls `buckets.invalidateQueue(name)`, which ranges the map and drops every `bucketKey` whose `queue` matches (regardless of incarnation), mirroring the `SetQueueAttributes` path above. The `CreateQueue` handler invokes the same call after a genuine create commit (the idempotent-return path skips it) so a same-name recreate that races with in-flight stale-meta traffic still resets the bucket. **Incarnation** participates in the key — `sqsQueueMeta.Incarnation` is set to `lastGen + 1` at `CreateQueue` time and is *preserved* across `PurgeQueue` and `SetQueueAttributes` (the read-modify-write on the meta record carries it forward). A `DeleteQueue`+`CreateQueue` cycle therefore lands the new incarnation at a different `bucketKey` and starts from a fresh full bucket regardless of any per-process cache the previous incarnation left behind on this or any other node. The two mechanisms are complementary — `invalidateQueue` is the cheap hot-path optimisation that keeps the in-memory map small, and the incarnation-keyed structure is the cross-leader correctness guarantee.
-**Why Incarnation, not Generation** (Codex P2 on PR #664 round 9): an earlier draft of this design used `sqsQueueMeta.Generation` as the bucket-key discriminator. `Generation` bumps on every `CreateQueue` *and* on every `PurgeQueue` (because message keys are prefixed with the generation, so a purge needs a new prefix to make the old data unreachable). Keying the throttle bucket by `Generation` would therefore re-key the bucket on every purge — letting any caller authorised to call `PurgeQueue` reset the rate limiter to a fresh full bucket once per purge. The 60-second AWS-spec rate limit on `PurgeQueue` bounds the bypass but does not eliminate it. `Incarnation` solves this by isolating the "queue identity changed" semantics (DeleteQueue+CreateQueue cycle) from the "data prefix changed" semantics (any of Create / Delete / Purge), and the throttle layer keys only on the former.
+**Why Incarnation, not Generation** (P2 review finding on PR #664 round 9): an earlier draft of this design used `sqsQueueMeta.Generation` as the bucket-key discriminator. `Generation` bumps on every `CreateQueue` *and* on every `PurgeQueue` (because message keys are prefixed with the generation, so a purge needs a new prefix to make the old data unreachable). Keying the throttle bucket by `Generation` would therefore re-key the bucket on every purge — letting any caller authorised to call `PurgeQueue` reset the rate limiter to a fresh full bucket once per purge. The 60-second AWS-spec rate limit on `PurgeQueue` bounds the bypass but does not eliminate it. `Incarnation` solves this by isolating the "queue identity changed" semantics (DeleteQueue+CreateQueue cycle) from the "data prefix changed" semantics (any of Create / Delete / Purge), and the throttle layer keys only on the former.
-The bucket map is per-process. There is no Raft replication of bucket state. The behaviour at the moment a node assumes leadership of a queue depends on which of three failover paths fires (Claude low on PR #664 round 7 caught the over-broad earlier wording):
+The bucket map is per-process. There is no Raft replication of bucket state. The behaviour at the moment a node assumes leadership of a queue depends on which of three failover paths fires (a low-severity review finding on PR #664 round 7 caught the over-broad earlier wording):
1. **First-time leader** (the most common case for a fresh process): no cached bucket, `loadOrInit` misses, the charge path mints a fresh bucket at full capacity.
2. **Re-elected node, throttle config changed during the prior leader's term**: `loadOrInit` finds the cached bucket but its `capacity`/`refillRate` no longer match the freshly-loaded meta, so the reconciliation path evicts the stale bucket and mints a fresh full-capacity replacement.
@@ -161,14 +161,14 @@ Setting `SendCapacity = 100, SendRefillPerSecond = 50` means: bursts up to 100 `
`Default*` fields catch any action not covered by an action-specific pair (so a future `PurgeQueue` rate limit costs nothing once defaults are wired).
-**Config-field → bucket-action mapping** (Codex P1 on PR #664 sixth-round Codex review): the JSON config field-name prefixes use short forms (`Send*`, `Recv*`, `Default*`) but the in-memory `bucketKey.action` from §3.1 uses the canonical action vocabulary (`"Send"`, `"Receive"`, `"*"`). The mapping is fixed: `Send*` → `bucketKey{action:"Send"}`, `Recv*` → `bucketKey{action:"Receive"}`, `Default*` → `bucketKey{action:"*"}`. Cache invalidation paragraphs in §3.1 use the bucket-action vocabulary (the actual map keys). Use the config-field vocabulary when discussing the JSON contract (`SetQueueAttributes` payload, `GetQueueAttributes` response) and the bucket-action vocabulary when discussing the in-memory map. Implementation must apply this mapping when looking up buckets after a `SetQueueAttributes` commit.
+**Config-field → bucket-action mapping** (P1 review finding on PR #664 sixth-round review): the JSON config field-name prefixes use short forms (`Send*`, `Recv*`, `Default*`) but the in-memory `bucketKey.action` from §3.1 uses the canonical action vocabulary (`"Send"`, `"Receive"`, `"*"`). The mapping is fixed: `Send*` → `bucketKey{action:"Send"}`, `Recv*` → `bucketKey{action:"Receive"}`, `Default*` → `bucketKey{action:"*"}`. Cache invalidation paragraphs in §3.1 use the bucket-action vocabulary (the actual map keys). Use the config-field vocabulary when discussing the JSON contract (`SetQueueAttributes` payload, `GetQueueAttributes` response) and the bucket-action vocabulary when discussing the in-memory map. Implementation must apply this mapping when looking up buckets after a `SetQueueAttributes` commit.
The `SetQueueAttributes` validator enforces:
- All four `Send*` / `Recv*` fields must be either both zero (disabled) or both positive.
- Capacity ≥ refill (otherwise the bucket can never burst above the steady state).
- A hard ceiling per queue (e.g. 100,000 RPS) so a typo (`SendCapacity = 1e9`) does not silently mean "no limit at all" but rejects with `InvalidAttributeValue`.
-- **Capacity ≥ max single-request charge** (Codex P1 on PR #664 sixth-round review). Per the §3.3 charging table, a `SendMessageBatch` charges up to 10 from the Send bucket and `DeleteMessageBatch` charges up to 10 from the Recv bucket (AWS caps both at 10 entries). Therefore: when `SendCapacity > 0` it must also be `≥ 10`, and when `RecvCapacity > 0` it must also be `≥ 10`. Without this rule, a queue configured with `SendCapacity = 5` enters a permanently unserviceable state for full batches — the bucket can never accumulate the 10 tokens a `SendMessageBatch(len=10)` requires, every full batch is rejected with `Throttling`, and `Retry-After` (§3.4) keeps reporting "wait N seconds" forever with no recovery path short of re-running `SetQueueAttributes`. The validator rejects with `InvalidAttributeValue` and an explicit message naming the per-bucket minimum so the operator sees the cause immediately. **`Default*` requires the same floor**: `resolveActionConfig` in `adapter/sqs_throttle.go` falls Send and Receive traffic through to the Default bucket whenever the dedicated `Send*` / `Recv*` pair is unset, so a `SendMessageBatch` or `DeleteMessageBatch` request can charge the Default bucket. A `DefaultCapacity < 10` therefore creates the same permanently-unserviceable-batch trap as `SendCapacity < 10`. (Earlier drafts of this proposal exempted `Default*` on the assumption that the catch-all set had no batch verb in scope; that was incorrect — the fall-through means batch verbs do hit Default*. The implementation passes `requireBatchCapacity = true` to `validateThrottlePair` for all three action sets — see Codex P1 on PR #679 round 5.)
+- **Capacity >= max single-request charge** (P1 review finding on PR #664 sixth-round review). Per the §3.3 charging table, a `SendMessageBatch` charges up to 10 from the Send bucket and `DeleteMessageBatch` charges up to 10 from the Recv bucket (AWS caps both at 10 entries). Therefore: when `SendCapacity > 0` it must also be `>= 10`, and when `RecvCapacity > 0` it must also be `>= 10`. Without this rule, a queue configured with `SendCapacity = 5` enters a permanently unserviceable state for full batches — the bucket can never accumulate the 10 tokens a `SendMessageBatch(len=10)` requires, every full batch is rejected with `Throttling`, and `Retry-After` (§3.4) keeps reporting "wait N seconds" forever with no recovery path short of re-running `SetQueueAttributes`. The validator rejects with `InvalidAttributeValue` and an explicit message naming the per-bucket minimum so the operator sees the cause immediately. **`Default*` requires the same floor**: `resolveActionConfig` in `adapter/sqs_throttle.go` falls Send and Receive traffic through to the Default bucket whenever the dedicated `Send*` / `Recv*` pair is unset, so a `SendMessageBatch` or `DeleteMessageBatch` request can charge the Default bucket. A `DefaultCapacity < 10` therefore creates the same permanently-unserviceable-batch trap as `SendCapacity < 10`. (Earlier drafts of this proposal exempted `Default*` on the assumption that the catch-all set had no batch verb in scope; that was incorrect — the fall-through means batch verbs do hit Default*. The implementation passes `requireBatchCapacity = true` to `validateThrottlePair` for all three action sets — see a P1 review finding on PR #679 round 5.)
### 3.3 Charging model
@@ -191,9 +191,9 @@ On rejection:
| Protocol | Response |
|---|---|
| JSON | HTTP 400, body `{"__type":"Throttling","message":"Rate exceeded for queue '' action ''"}`, header `x-amzn-ErrorType: Throttling`, header `Retry-After: ` (computed per below) |
-| Query | HTTP 400, body `SenderThrottling......`, headers as above |
+| Query | Not implemented for throttled message verbs in this lifecycle update. The current query dispatcher wires catalog verbs only and returns `NotImplementedYet` for message actions; when query message handlers land, they should mirror the JSON throttling code with `Throttling` in the XML error body. |
-`Retry-After` is computed from the *actual* refill rate AND the *requested* token count so neither slow refill nor large batches cause a busy-loop of premature retries (two consecutive Claude reviews on PR #664 caught both: first the `Retry-After: 1` constant lying for sub-1-RPS refill — `SendRefillPerSecond = 0.1` needs 10 s for the next token; then the formula's hardcoded numerator `1.0` lying for batch verbs that charge >1 token — a `SendMessageBatch` of 10 against `refillRate = 1.0` and 0 tokens needs 10 s, not 1):
+`Retry-After` is computed from the *actual* refill rate AND the *requested* token count so neither slow refill nor large batches cause a busy-loop of premature retries (two consecutive review findings on PR #664 caught both: first the `Retry-After: 1` constant lying for sub-1-RPS refill — `SendRefillPerSecond = 0.1` needs 10 s for the next token; then the formula's hardcoded numerator `1.0` lying for batch verbs that charge >1 token — a `SendMessageBatch` of 10 against `refillRate = 1.0` and 0 tokens needs 10 s, not 1):
```text
needed := float64(requestedCount) - currentTokens
@@ -213,13 +213,14 @@ The minimum-1 floor matches `Retry-After`'s integer-second granularity (HTTP/1.1
| File | Change |
|---|---|
-| `adapter/sqs_throttle.go` (new) | `bucketStore`, `tokenBucket`, charging helper. ~250 lines. |
+| `adapter/sqs_throttle.go` | `bucketStore`, `tokenBucket`, charging helper. |
| `adapter/sqs_catalog.go` | Add `Throttle` field to `sqsQueueMeta`. Extend `applyAttributes` with the new `Throttle*` attribute names. Render the four Throttle fields in `queueMetaToAttributes` so `GetQueueAttributes("All")` surfaces them. |
-| `adapter/sqs.go` | After `authorizeSQSRequest`, call `bucketStore.charge(queueName, action, count)`. On reject, write the `Throttling` envelope and return. |
-| `adapter/sqs_throttle_test.go` (new) | Unit tests for bucket math (edge cases: idle drift, burst, partial refill, batch over-charge, default-off). ~300 lines. |
-| `adapter/sqs_throttle_integration_test.go` (new) | End-to-end: configure a queue with low limits, send N messages back-to-back, confirm the (N+1)th gets `Throttling` with `Retry-After`. ~150 lines. |
-| `monitoring/registry.go` | New counter `sqs_throttled_requests_total{queue, action}` and new **gauge** `sqs_throttle_tokens_remaining{queue, action}`. (Codex P2 on PR #664: tokens go up *and* down so a counter is the wrong instrument.) |
-| `docs/design/2026_04_24_partial_sqs_compatible_adapter.md` §16.5 | Status update once this lands: TODO → Landed. |
+| `adapter/sqs.go` | After `authorizeSQSRequest`, call `bucketStore.charge(queueName, action, count)`. On reject, write the `Throttling` envelope and return. Also forwards each configured bucket decision to the SQS throttle metrics observer. |
+| `adapter/sqs_throttle_test.go` | Unit tests for bucket math, observer emission, disabled-bucket metric cleanup, idle drift, burst, partial refill, batch over-charge, and default-off. |
+| `adapter/sqs_throttle_integration_test.go` | End-to-end: configure a queue with low limits, send N messages back-to-back, confirm the (N+1)th gets `Throttling` with `Retry-After`. |
+| `monitoring/sqs.go`, `monitoring/registry.go` | Register the `elastickv_sqs_throttled_requests_total{queue, action}` counter and `elastickv_sqs_throttle_tokens_remaining{queue, action}` gauge. Queue-label cardinality uses the same capped `_other` fallback as the existing SQS metrics. |
+| `monitoring/sqs_test.go` | Pins public registration of both throttle metrics, independent queue-label budgets for throttle counters vs partition counters, short-lived queue cleanup, disabled action cleanup, and action-aware `_other` gauge cleanup. |
+| `docs/design/2026_04_24_proposed_sqs_compatible_adapter.md` §14 | Phase 3 per-queue throttling item marked landed with a link to this design. |
### 4.2 OCC interaction
@@ -229,9 +230,9 @@ Throttling sits *outside* the OCC transaction — a rejected request never touch
Each queue is owned by exactly one shard (queue-per-shard routing in `kv/shard_router.go`). The leader of that shard owns the bucket. A request that lands on a follower is forwarded by `proxyToLeader` *before* the bucket check, so the bucket is always evaluated by the leader that is also doing the OCC dispatch — no risk of a follower checking against a stale bucket and the leader committing without checking.
-Once Phase 3.D (split-queue FIFO) lands, a single queue may span multiple shards. At that point each *partition* gets its own bucket, **keyed by `(queueName, partitionID)`** — not by `MessageGroupId`. `MessageGroupId` is the *input* to `partitionFor`; using it directly as the bucket key would create one bucket per unique group value (unbounded, attacker-amplifiable map size, and hot groups would never share a budget). `partitionID` is bounded by `PartitionCount` so the worst-case bucket count per queue is tiny. The throttle proposal is forward-compatible: the bucket lookup key changes from `queueName` to `(queueName, partitionID)`, and the `bucketKey` struct in §3.1 grows a `partition uint32` field. Documented in §11. (Claude P1 on PR #664 caught the misnomer.)
+Split-queue FIFO has landed, but the throttle implementation still uses one aggregate bucket per `(queueName, action, incarnation)`. Partitioned queues therefore share the configured `SendCapacity` / `RecvCapacity` / `DefaultCapacity` across the logical queue; the effective queue throughput is the configured capacity, not `PartitionCount` times that value. This matches the current dispatch path: `SendMessage` charges the queue-level bucket after loading queue metadata and before partition-specific routing.
-**Budget semantics per partition:** each partition's bucket gets the *full* configured `SendCapacity` / `RecvCapacity` / `DefaultCapacity`. The effective aggregate throughput of an N-partition queue is therefore N × the configured per-partition limit. This is intentional and analogous to how AWS High Throughput FIFO multiplies throughput by partition count; operators sizing the throttle should treat `SendCapacity` as the *per-partition* budget. A shared queue-level budget (divided across partitions) would require cross-shard coordination on every `SendMessage` — an extra Raft round-trip per call, defeating the point of partitioning. If per-queue aggregate throttling is needed after Phase 3.D lands, a new `SendCapacityTotal` attribute could be added that gets divided by `PartitionCount` at config time and stored as the per-partition capacity; that design is out of scope for this proposal.
+Per-partition buckets remain future work. That change must move the charge point to a path that already knows `partitionFor(meta, MessageGroupId)`, extend `bucketKey` with a bounded `partition uint32` field, and audit the send/receive/batch callers so every throttled action uses the same bucket semantics. `MessageGroupId` itself must never be used as the bucket key because it is unbounded and caller-controlled.
---
@@ -272,7 +273,7 @@ Strict-validation SDKs that reject unknown attribute names will reject `Throttle
4. **Configuration round-trip**: `SetQueueAttributes` with throttle config → `GetQueueAttributes` returns the same values; an unknown `Throttle*` attribute name is rejected with 400 `InvalidAttributeName` (matching AWS behaviour for unrecognised attributes).
-5. **Cross-protocol parity**: throttled JSON and Query requests both surface `Throttling` (different envelope, same code).
+5. **JSON protocol coverage**: throttled JSON requests surface `Throttling` with `Retry-After`. Query/XML parity is a follow-up gated on wiring the query message handlers to the same throttle path.
6. **Failover behaviour** (3-node cluster): kill the current leader after 3 messages, confirm the next leader starts the bucket fresh and accepts up to capacity again. Log line records the failover so operators can correlate.
@@ -284,10 +285,12 @@ Strict-validation SDKs that reject unknown attribute names will reject `Throttle
No new flags. Limits are per-queue, set via `SetQueueAttributes`. Defaults are zero (disabled).
-Two new Prometheus instruments (Section 4.1) expose the throttling activity:
+Throttle outcomes are visible in two places:
-- `sqs_throttled_requests_total{queue, action}` — **counter**. Use `rate(...)` per queue in Grafana to spot the noisy tenant.
-- `sqs_throttle_tokens_remaining{queue, action}` — **gauge** (Codex P2 on PR #664: token budgets go up *and* down over time, so a counter would mask the depletion that operators most need to see). Sample directly; trending toward zero is the early warning sign.
+1. The HTTP response: rejected JSON-path message requests return `Throttling` with `Retry-After`.
+2. Prometheus: `elastickv_sqs_throttled_requests_total{queue, action}` increments for rejected requests, and `elastickv_sqs_throttle_tokens_remaining{queue, action}` records the remaining token balance after each configured bucket decision. `action` is the effective bucket action: `send`, `receive`, or `default`.
+
+No server-side throttle access log is required for this lifecycle; deployments that need request-level forensic logs can add an HTTP access/audit layer independently of the rate-limit correctness surface.
---
@@ -323,10 +326,10 @@ Every `charge` proposes a bucket update through the FSM. **Rejected**: an extra
| Phase | Content |
|---|---|
-| 1 | Doc lands (this PR). No code yet. Operators have time to comment. |
-| 2 | Implementation PR per §4.1. Default-off; existing queues unaffected. |
-| 3 | Operators opt in per queue via `SetQueueAttributes`. Monitor `sqs_throttled_requests_total` for false positives. |
-| 4 | Once stable, the partial doc's TODO list moves 3.C from TODO to Landed. |
+| 1 | Default-off implementation and lifecycle documentation have landed; existing queues remain unthrottled until configured. |
+| 2 | Operators opt in per queue via `SetQueueAttributes`. Watch caller-visible `Throttling` responses and `Retry-After` for false positives. |
+| 3 | Prometheus exports `elastickv_sqs_throttled_requests_total{queue, action}` and `elastickv_sqs_throttle_tokens_remaining{queue, action}` with bounded queue-label cardinality. |
+| 4 | Parent SQS adapter roadmap marks Phase 3.C as landed with a link to this design. |
---
diff --git a/docs/design/2026_06_23_proposed_scaling_roadmap.md b/docs/design/2026_06_23_proposed_scaling_roadmap.md
index 2bd704690..625f0e37c 100644
--- a/docs/design/2026_06_23_proposed_scaling_roadmap.md
+++ b/docs/design/2026_06_23_proposed_scaling_roadmap.md
@@ -289,7 +289,7 @@ memory each group's private cache/memtable pins.
command worker pool, optional Raft-thread pinning, per-client admission,
XREAD O(N)→O(new)), S3 PUT admission control
(`2026_04_25_proposed_s3_admission_control.md`), SQS per-queue throttling
- (`2026_04_26_proposed_sqs_per_queue_throttling.md`). These bound *one
+ (`2026_04_26_implemented_sqs_per_queue_throttling.md`). These bound *one
workload's* impact so it cannot starve the shared runtime / Raft control
plane; they scale a deployment by making it predictable under adversarial or
unbalanced load rather than by raising raw capacity.
diff --git a/main.go b/main.go
index 894c7c4b1..92fd9cf0e 100644
--- a/main.go
+++ b/main.go
@@ -1506,7 +1506,9 @@ func startSQSDepthObserver(ctx context.Context, reg *monitoring.Registry, sqsSer
// expects []monitoring.SQSQueueDepth). Same shape both sides; the
// loop is a fixed-size copy.
type sqsDepthSourceAdapter struct {
- inner *adapter.SQSServer
+ inner interface {
+ SnapshotQueueDepths(context.Context) ([]adapter.SQSQueueDepth, bool)
+ }
}
func (a sqsDepthSourceAdapter) SnapshotQueueDepths(ctx context.Context) ([]monitoring.SQSQueueDepth, bool) {
@@ -1521,7 +1523,7 @@ func (a sqsDepthSourceAdapter) SnapshotQueueDepths(ctx context.Context) ([]monit
// existing gauges alone on a transient scan failure.
return nil, false
}
- if len(snaps) == 0 {
+ if snaps == nil {
return nil, true
}
out := make([]monitoring.SQSQueueDepth, len(snaps))
diff --git a/main_sqs.go b/main_sqs.go
index 85b9fd488..fc2a43fcd 100644
--- a/main_sqs.go
+++ b/main_sqs.go
@@ -46,6 +46,10 @@ func startSQSServer(
closeSQSListenerOnError(sqsL, sqsAddr)
return nil, err
}
+ var throttleObserver adapter.SQSThrottleObserver
+ if o, ok := partitionObserver.(adapter.SQSThrottleObserver); ok {
+ throttleObserver = o
+ }
sqsServer := adapter.NewSQSServer(
sqsL,
shardStore,
@@ -55,6 +59,7 @@ func startSQSServer(
adapter.WithSQSStaticCredentials(staticCreds),
adapter.WithSQSPartitionResolver(partitionResolver),
adapter.WithSQSPartitionObserver(partitionObserver),
+ adapter.WithSQSThrottleObserver(throttleObserver),
)
// Two-goroutine shutdown pattern mirrors startS3Server: one goroutine waits
// on either ctx.Done() or Run completion to call Stop, the other runs the
diff --git a/main_sqs_depth_observer_test.go b/main_sqs_depth_observer_test.go
new file mode 100644
index 000000000..f5457cb38
--- /dev/null
+++ b/main_sqs_depth_observer_test.go
@@ -0,0 +1,33 @@
+package main
+
+import (
+ "context"
+ "testing"
+
+ "github.com/bootjp/elastickv/adapter"
+ "github.com/stretchr/testify/require"
+)
+
+type fakeSQSDepthBridgeSource struct {
+ snaps []adapter.SQSQueueDepth
+ ok bool
+}
+
+func (f fakeSQSDepthBridgeSource) SnapshotQueueDepths(context.Context) ([]adapter.SQSQueueDepth, bool) {
+ return f.snaps, f.ok
+}
+
+func TestSQSDepthSourceAdapterPreservesNilVsEmptySnapshots(t *testing.T) {
+ t.Parallel()
+
+ got, ok := (sqsDepthSourceAdapter{inner: fakeSQSDepthBridgeSource{snaps: nil, ok: true}}).
+ SnapshotQueueDepths(context.Background())
+ require.True(t, ok)
+ require.Nil(t, got, "nil successful snapshots must remain nil for follower/step-down cleanup")
+
+ got, ok = (sqsDepthSourceAdapter{inner: fakeSQSDepthBridgeSource{snaps: []adapter.SQSQueueDepth{}, ok: true}}).
+ SnapshotQueueDepths(context.Background())
+ require.True(t, ok)
+ require.NotNil(t, got, "non-nil empty leader snapshots must survive the adapter bridge")
+ require.Empty(t, got)
+}
diff --git a/monitoring/sqs.go b/monitoring/sqs.go
index aa1d0a2b4..353b7405f 100644
--- a/monitoring/sqs.go
+++ b/monitoring/sqs.go
@@ -16,6 +16,10 @@ const (
SQSPartitionActionReceive = "receive"
SQSPartitionActionDelete = "delete"
+ SQSThrottleActionSend = "send"
+ SQSThrottleActionReceive = "receive"
+ SQSThrottleActionDefault = "default"
+
// sqsMaxTrackedQueues caps the number of distinct queue names
// the metrics layer will emit a per-(queue, partition, action)
// series for. Any queue beyond this cap collapses to the
@@ -45,6 +49,15 @@ type SQSPartitionObserver interface {
ObservePartitionMessage(queue string, partition uint32, action string)
}
+// SQSThrottleObserver records per-queue throttle decisions. The SQS
+// adapter calls it after a configured token-bucket charge so operators
+// can observe both rejected requests and the live token balance.
+type SQSThrottleObserver interface {
+ ObserveThrottleDecision(queue string, action string, tokensRemaining float64, throttled bool)
+ ForgetThrottleAction(queue string, action string)
+ SyncThrottleActions(queue string, enabledActions []string)
+}
+
// SQSDepthSource is the contract a per-tick queue-depth source must
// satisfy. Implemented by *adapter.SQSServer; SQSObserver.Start
// calls SnapshotQueueDepths on every interval and writes the
@@ -57,12 +70,16 @@ type SQSPartitionObserver interface {
//
// Two distinct empty-snapshot states, signalled by ok:
//
-// - ok=true with an empty/nil slice — "this source legitimately
-// has no queues this tick". Triggers when the node is a
-// follower (leader-only emission) or the leader genuinely has
-// zero queues configured. The observer diffs against the
-// previous tick and ForgetQueue's any queue that disappeared
-// so a former leader's gauges are cleared on step-down.
+// - ok=true with a nil slice — "this source is not emitting this tick".
+// The adapter returns this when the node is a follower (leader-only
+// emission). The observer clears both depth and token gauges so a former
+// leader does not keep exporting stale SQS series after step-down.
+//
+// - ok=true with a non-nil slice, possibly empty — leader, scrape OK.
+// The observer writes the returned queue gauges and diffs against the
+// previous tick. A queue omitted from a non-nil snapshot can be a
+// per-queue scan miss, so throttle token gauges for depth-seen queues are
+// preserved until a config/delete path explicitly removes them.
//
// - ok=false (regardless of the slice contents) — "the source
// could not produce a snapshot this tick" (transient catalog-
@@ -92,11 +109,11 @@ type SQSQueueDepth struct {
// SQSMetrics owns the Prometheus collectors for the SQS adapter.
// Mirrors DynamoDBMetrics' shape: per-Registry instance, label-
-// cardinality-bounded by sqsMaxTrackedQueues, and split between
-// counters (HT-FIFO partition activity) and gauges (queue depth).
+// cardinality-bounded by sqsMaxTrackedQueues, and split by metric
+// family so unrelated SQS signals cannot consume each other's
+// queue-label budget.
//
-// The cardinality budget is split into two independent maps because
-// the two metrics have different deletion semantics:
+// Counter and gauge families have different deletion semantics:
//
// - partitionMessages is a CounterVec. Counters are cumulative;
// deleting a series throws away its observed-since-process-start
@@ -104,14 +121,24 @@ type SQSQueueDepth struct {
// budget therefore only ever grows — once a queue is admitted to
// trackedCounterQueues it stays admitted, and ForgetQueue does
// NOT touch this map.
+// - throttledRequests is also cumulative, but uses its own
+// trackedThrottleCounterQueues admission map. A burst of
+// rejected non-partitioned queues must not force later HT-FIFO
+// partition counters into _other, and partition traffic must not
+// hide throttle rejects.
// - queueDepth is a GaugeVec. Gauges have no cumulative state, so
// DeleteLabelValues is safe (and necessary, otherwise a deleted
// queue keeps reporting a frozen backlog on the dashboard).
// ForgetQueue both removes the gauge series and frees the
// queue's slot in trackedDepthQueues so a churn-heavy deployment
// can reuse the budget.
+// - throttleTokens is also a GaugeVec, but decisions are event-
+// driven and action-labelled. It has a separate queue/action
+// lifetime from the depth scraper: throttle config changes or
+// throttle-only cleanup drop token gauges, but transient depth
+// scan misses must not delete them.
//
-// Sharing one map across the two metrics regresses the counter cap:
+// Sharing one map across counters and gauges regresses the counter cap:
// ForgetQueue would free a slot, a new queue would be admitted, and
// the previous queue's counter series would still occupy a real-name
// label in Prometheus — letting cardinality grow without bound under
@@ -120,10 +147,17 @@ type SQSQueueDepth struct {
type SQSMetrics struct {
partitionMessages *prometheus.CounterVec
queueDepth *prometheus.GaugeVec
+ throttledRequests *prometheus.CounterVec
+ throttleTokens *prometheus.GaugeVec
- mu sync.Mutex
- trackedCounterQueues map[string]struct{}
- trackedDepthQueues map[string]struct{}
+ mu sync.Mutex
+ trackedCounterQueues map[string]struct{}
+ trackedThrottleCounterQueues map[string]struct{}
+ trackedDepthQueues map[string]struct{}
+ trackedThrottleGaugeQueues map[string]map[string]struct{}
+ throttleGaugeQueueSeq map[string]uint64
+ throttleGaugeActionSeq map[string]map[string]uint64
+ throttleGaugeSeq uint64
// overflowDepthQueues is the set of queue names whose depth
// gauge is currently collapsed onto the shared sqsQueueOverflow
// label. Tracked separately from trackedDepthQueues (which
@@ -138,7 +172,8 @@ type SQSMetrics struct {
// produce phantom data — it correctly reflects the cumulative
// count of all overflow operations, even from queues that no
// longer exist.
- overflowDepthQueues map[string]struct{}
+ overflowDepthQueues map[string]struct{}
+ overflowThrottleGaugeQueues map[string]map[string]struct{}
}
// SQS depth gauge state-label values. Stable so dashboards / alerts
@@ -165,12 +200,33 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics {
},
[]string{"queue", "state"},
),
- trackedCounterQueues: map[string]struct{}{},
- trackedDepthQueues: map[string]struct{}{},
- overflowDepthQueues: map[string]struct{}{},
+ throttledRequests: prometheus.NewCounterVec(
+ prometheus.CounterOpts{
+ Name: "elastickv_sqs_throttled_requests_total",
+ Help: "Total SQS requests rejected by per-queue throttling, labelled by queue and effective throttle bucket action.",
+ },
+ []string{"queue", "action"},
+ ),
+ throttleTokens: prometheus.NewGaugeVec(
+ prometheus.GaugeOpts{
+ Name: "elastickv_sqs_throttle_tokens_remaining",
+ Help: "Remaining tokens after the most recent SQS per-queue throttle decision, labelled by queue and effective throttle bucket action.",
+ },
+ []string{"queue", "action"},
+ ),
+ trackedCounterQueues: map[string]struct{}{},
+ trackedThrottleCounterQueues: map[string]struct{}{},
+ trackedDepthQueues: map[string]struct{}{},
+ trackedThrottleGaugeQueues: map[string]map[string]struct{}{},
+ throttleGaugeQueueSeq: map[string]uint64{},
+ throttleGaugeActionSeq: map[string]map[string]uint64{},
+ overflowDepthQueues: map[string]struct{}{},
+ overflowThrottleGaugeQueues: map[string]map[string]struct{}{},
}
registerer.MustRegister(m.partitionMessages)
registerer.MustRegister(m.queueDepth)
+ registerer.MustRegister(m.throttledRequests)
+ registerer.MustRegister(m.throttleTokens)
return m
}
@@ -220,11 +276,126 @@ func (m *SQSMetrics) ObserveQueueDepth(queue string, visible, notVisible, delaye
m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateDelayed).Set(float64(max(int64(0), delayed)))
}
-// ForgetQueue drops the three gauge series for a queue and frees
-// its slot in the depth-side cardinality budget so a long-running
-// deployment that regularly creates and deletes queues (CI
-// workloads, ephemeral per-job queues) doesn't permanently wedge
-// the 512-entry depth budget.
+// ObserveThrottleDecision implements SQSThrottleObserver. It updates
+// the current token-balance gauge for every configured bucket decision
+// and increments the rejected-request counter only when throttled=true.
+func (m *SQSMetrics) ObserveThrottleDecision(queue string, action string, tokensRemaining float64, throttled bool) {
+ if m == nil || queue == "" {
+ return
+ }
+ if !sqsValidThrottleAction(action) {
+ return
+ }
+ if tokensRemaining < 0 {
+ tokensRemaining = 0
+ }
+ if throttled {
+ queueCounterLabel := m.admitForThrottleCounterBudget(queue)
+ m.throttledRequests.WithLabelValues(queueCounterLabel, action).Inc()
+ }
+ m.mu.Lock()
+ queueGaugeLabel := m.admitForThrottleGaugeBudgetLocked(queue, action)
+ m.throttleTokens.WithLabelValues(queueGaugeLabel, action).Set(tokensRemaining)
+ m.mu.Unlock()
+}
+
+// ObserveThrottleRejection increments only the rejected-request counter. The
+// adapter uses it when a stale pre-invalidation request still returns
+// Throttling to the client but must not recreate a token-balance gauge.
+func (m *SQSMetrics) ObserveThrottleRejection(queue string, action string) {
+ if m == nil || queue == "" {
+ return
+ }
+ if !sqsValidThrottleAction(action) {
+ return
+ }
+ queueCounterLabel := m.admitForThrottleCounterBudget(queue)
+ m.throttledRequests.WithLabelValues(queueCounterLabel, action).Inc()
+}
+
+// ForgetThrottleAction removes the token-balance gauge for one queue/action
+// whose throttle bucket is no longer configured. Throttled-request counters
+// are cumulative and intentionally survive.
+func (m *SQSMetrics) ForgetThrottleAction(queue string, action string) {
+ if m == nil || queue == "" {
+ return
+ }
+ if !sqsValidThrottleAction(action) {
+ return
+ }
+ m.mu.Lock()
+ forget := m.forgetThrottleActionLocked(queue, action)
+ if forget.tracked {
+ m.dropThrottleGaugeActionFor(queue, action)
+ }
+ if forget.overflowSetEmpty {
+ m.dropThrottleGaugeActionFor(sqsQueueOverflow, action)
+ }
+ m.mu.Unlock()
+}
+
+// ForgetThrottleActionBefore removes a token-balance gauge only when that
+// specific queue/action was not refreshed after cutoff. It lets config-change
+// cleanup reset stale gauges without erasing a fresh post-change request that
+// raced between bucket invalidation and metrics reconciliation.
+func (m *SQSMetrics) ForgetThrottleActionBefore(queue string, action string, cutoff uint64) {
+ if m == nil || queue == "" {
+ return
+ }
+ if !sqsValidThrottleAction(action) {
+ return
+ }
+ m.mu.Lock()
+ if actionSeqs := m.throttleGaugeActionSeq[queue]; actionSeqs != nil {
+ if seq := actionSeqs[action]; seq > cutoff {
+ m.mu.Unlock()
+ return
+ }
+ }
+ forget := m.forgetThrottleActionLocked(queue, action)
+ if forget.tracked {
+ m.dropThrottleGaugeActionFor(queue, action)
+ }
+ if forget.overflowSetEmpty {
+ m.dropThrottleGaugeActionFor(sqsQueueOverflow, action)
+ }
+ m.mu.Unlock()
+}
+
+// SyncThrottleActions reconciles token-balance gauges after a queue's throttle
+// configuration changes, even when no later request touches the disabled
+// bucket. enabledActions is the configured metric-action set; omitted actions
+// have their gauges removed. Counters intentionally survive.
+func (m *SQSMetrics) SyncThrottleActions(queue string, enabledActions []string) {
+ if m == nil || queue == "" {
+ return
+ }
+ enabled := make(map[string]struct{}, len(enabledActions))
+ for _, action := range enabledActions {
+ if sqsValidThrottleAction(action) {
+ enabled[action] = struct{}{}
+ }
+ }
+ m.mu.Lock()
+ for _, action := range []string{SQSThrottleActionSend, SQSThrottleActionReceive, SQSThrottleActionDefault} {
+ if _, ok := enabled[action]; ok {
+ continue
+ }
+ forget := m.forgetThrottleActionLocked(queue, action)
+ if forget.tracked {
+ m.dropThrottleGaugeActionFor(queue, action)
+ }
+ if forget.overflowSetEmpty {
+ m.dropThrottleGaugeActionFor(sqsQueueOverflow, action)
+ }
+ }
+ m.mu.Unlock()
+}
+
+// ForgetQueue drops the three depth gauge series for a queue and frees its slot
+// in the depth-side cardinality budget so a long-running deployment that
+// regularly creates and deletes queues (CI workloads, ephemeral per-job queues)
+// doesn't permanently wedge the 512-entry depth budget.
//
// Three cases, by membership at call time:
//
@@ -254,22 +425,51 @@ func (m *SQSMetrics) ObserveQueueDepth(queue string, visible, notVisible, delaye
// have since been deleted.
//
// Caller-audit per the standing semantic-change rule: only
-// SQSObserver.observeOnce calls this (registry plumbing aside),
-// and it's invoked exactly when a queue is observed in the
-// previous tick but not the current one. The caller's contract —
-// "drop gauges for a queue that disappeared so dashboards don't
-// show frozen backlog" — is preserved AND extended consistently:
-// overflow queues, previously a silent no-op, now also stop
-// pinning the shared gauge once the last one disappears. The
-// narrowed-but-still-correct scope (counter budget never
-// reclaimed) is invisible to observeOnce because the observer
-// only writes gauges; counters are fed via ObservePartitionMessage
-// from a different code path entirely.
+// SQSObserver.observeOnce calls this (registry plumbing aside), and it is invoked
+// exactly when a queue was observed in the previous tick but not the current one.
+// The caller's depth contract is preserved, including overflow ref-counting, but
+// throttle token gauges are intentionally out of scope: a per-queue depth scan
+// miss is not evidence that throttle config changed.
func (m *SQSMetrics) ForgetQueue(queue string) {
if m == nil || queue == "" {
return
}
m.mu.Lock()
+ depthForget := m.forgetDepthQueueLocked(queue)
+ m.mu.Unlock()
+ m.dropForgottenDepthQueue(queue, depthForget)
+}
+
+// ForgetThrottleQueue drops every token-balance gauge for queue and frees its
+// throttle-gauge budget slot. It is intentionally separate from ForgetQueue
+// because queue-depth snapshots can omit a live queue on a transient per-queue
+// scan error.
+func (m *SQSMetrics) ForgetThrottleQueue(queue string) {
+ if m == nil || queue == "" {
+ return
+ }
+ m.mu.Lock()
+ throttleForget := m.forgetThrottleQueueLocked(queue)
+ m.dropForgottenThrottleQueue(queue, throttleForget)
+ m.mu.Unlock()
+}
+
+type sqsGaugeForget struct {
+ tracked bool
+ overflowSetEmpty bool
+}
+
+type sqsThrottleGaugeForget struct {
+ tracked bool
+ overflowActions []string
+}
+
+type sqsThrottleActionForget struct {
+ tracked bool
+ overflowSetEmpty bool
+}
+
+func (m *SQSMetrics) forgetDepthQueueLocked(queue string) sqsGaugeForget {
_, tracked := m.trackedDepthQueues[queue]
if tracked {
delete(m.trackedDepthQueues, queue)
@@ -282,12 +482,134 @@ func (m *SQSMetrics) ForgetQueue(queue string) {
if overflow {
delete(m.overflowDepthQueues, queue)
}
- overflowSetEmpty := overflow && len(m.overflowDepthQueues) == 0
- m.mu.Unlock()
- if tracked {
+ return sqsGaugeForget{tracked: tracked, overflowSetEmpty: overflow && len(m.overflowDepthQueues) == 0}
+}
+
+func (m *SQSMetrics) forgetThrottleQueueLocked(queue string) sqsThrottleGaugeForget {
+ _, throttleTracked := m.trackedThrottleGaugeQueues[queue]
+ if throttleTracked {
+ delete(m.trackedThrottleGaugeQueues, queue)
+ }
+ delete(m.throttleGaugeQueueSeq, queue)
+ delete(m.throttleGaugeActionSeq, queue)
+ return sqsThrottleGaugeForget{
+ tracked: throttleTracked,
+ overflowActions: m.removeThrottleOverflowQueueLocked(queue),
+ }
+}
+
+func (m *SQSMetrics) forgetThrottleActionLocked(queue string, action string) sqsThrottleActionForget {
+ var forget sqsThrottleActionForget
+ if actions, ok := m.trackedThrottleGaugeQueues[queue]; ok {
+ if _, hasAction := actions[action]; hasAction {
+ delete(actions, action)
+ forget.tracked = true
+ }
+ if len(actions) == 0 {
+ delete(m.trackedThrottleGaugeQueues, queue)
+ }
+ }
+ if queues, ok := m.overflowThrottleGaugeQueues[action]; ok {
+ if _, hasQueue := queues[queue]; hasQueue {
+ delete(queues, queue)
+ if len(queues) == 0 {
+ delete(m.overflowThrottleGaugeQueues, action)
+ forget.overflowSetEmpty = true
+ }
+ }
+ }
+ if !m.throttleGaugeQueueTrackedLocked(queue) {
+ delete(m.throttleGaugeQueueSeq, queue)
+ }
+ if !m.throttleGaugeActionTrackedLocked(queue, action) {
+ m.deleteThrottleGaugeActionSeqLocked(queue, action)
+ }
+ return forget
+}
+
+func (m *SQSMetrics) removeThrottleOverflowQueueLocked(queue string) []string {
+ var emptyActions []string
+ for action, queues := range m.overflowThrottleGaugeQueues {
+ if _, ok := queues[queue]; !ok {
+ continue
+ }
+ delete(queues, queue)
+ if len(queues) == 0 {
+ delete(m.overflowThrottleGaugeQueues, action)
+ emptyActions = append(emptyActions, action)
+ }
+ }
+ if !m.throttleGaugeQueueTrackedLocked(queue) {
+ delete(m.throttleGaugeQueueSeq, queue)
+ }
+ delete(m.throttleGaugeActionSeq, queue)
+ return emptyActions
+}
+
+func (m *SQSMetrics) removeThrottleOverflowActionLocked(queue string, action string) bool {
+ queues, ok := m.overflowThrottleGaugeQueues[action]
+ if !ok {
+ return false
+ }
+ if _, ok := queues[queue]; !ok {
+ return false
+ }
+ delete(queues, queue)
+ empty := len(queues) == 0
+ if empty {
+ delete(m.overflowThrottleGaugeQueues, action)
+ }
+ if !m.throttleGaugeActionTrackedLocked(queue, action) {
+ m.deleteThrottleGaugeActionSeqLocked(queue, action)
+ }
+ if !m.throttleGaugeQueueTrackedLocked(queue) {
+ delete(m.throttleGaugeQueueSeq, queue)
+ }
+ return empty
+}
+
+func (m *SQSMetrics) throttleGaugeQueueTrackedLocked(queue string) bool {
+ if _, ok := m.trackedThrottleGaugeQueues[queue]; ok {
+ return true
+ }
+ for _, queues := range m.overflowThrottleGaugeQueues {
+ if _, ok := queues[queue]; ok {
+ return true
+ }
+ }
+ return false
+}
+
+func (m *SQSMetrics) throttleGaugeActionTrackedLocked(queue string, action string) bool {
+ if actions, ok := m.trackedThrottleGaugeQueues[queue]; ok {
+ if _, ok := actions[action]; ok {
+ return true
+ }
+ }
+ if queues, ok := m.overflowThrottleGaugeQueues[action]; ok {
+ if _, ok := queues[queue]; ok {
+ return true
+ }
+ }
+ return false
+}
+
+func (m *SQSMetrics) deleteThrottleGaugeActionSeqLocked(queue string, action string) {
+ actionSeqs := m.throttleGaugeActionSeq[queue]
+ if actionSeqs == nil {
+ return
+ }
+ delete(actionSeqs, action)
+ if len(actionSeqs) == 0 {
+ delete(m.throttleGaugeActionSeq, queue)
+ }
+}
+
+func (m *SQSMetrics) dropForgottenDepthQueue(queue string, forget sqsGaugeForget) {
+ if forget.tracked {
m.dropGaugeStatesFor(queue)
}
- if overflowSetEmpty {
+ if forget.overflowSetEmpty {
// Last overflow queue forgotten — drop the shared _other
// series so dashboards stop showing phantom backlog for
// queues that no longer exist. Safe because no remaining
@@ -297,6 +619,15 @@ func (m *SQSMetrics) ForgetQueue(queue string) {
}
}
+func (m *SQSMetrics) dropForgottenThrottleQueue(queue string, forget sqsThrottleGaugeForget) {
+ if forget.tracked {
+ m.dropThrottleGaugeActionsFor(queue)
+ }
+ for _, action := range forget.overflowActions {
+ m.dropThrottleGaugeActionFor(sqsQueueOverflow, action)
+ }
+}
+
// admitForCounterBudget returns the canonical label for queue: the
// real name when the counter budget has room, sqsQueueOverflow once
// it is saturated. The counter budget never shrinks (counter series
@@ -306,13 +637,23 @@ func (m *SQSMetrics) ForgetQueue(queue string) {
func (m *SQSMetrics) admitForCounterBudget(queue string) string {
m.mu.Lock()
defer m.mu.Unlock()
- if _, ok := m.trackedCounterQueues[queue]; ok {
+ return admitCounterQueueLocked(queue, m.trackedCounterQueues)
+}
+
+func (m *SQSMetrics) admitForThrottleCounterBudget(queue string) string {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return admitCounterQueueLocked(queue, m.trackedThrottleCounterQueues)
+}
+
+func admitCounterQueueLocked(queue string, tracked map[string]struct{}) string {
+ if _, ok := tracked[queue]; ok {
return queue
}
- if len(m.trackedCounterQueues) >= sqsMaxTrackedQueues {
+ if len(tracked) >= sqsMaxTrackedQueues {
return sqsQueueOverflow
}
- m.trackedCounterQueues[queue] = struct{}{}
+ tracked[queue] = struct{}{}
return queue
}
@@ -360,6 +701,87 @@ func (m *SQSMetrics) admitForDepthBudget(queue string) string {
return queue
}
+// admitForThrottleGaugeBudgetLocked mirrors admitForDepthBudget for the
+// event-driven throttle-token gauge. It is separate from queue depth because a
+// queue can emit throttle decisions without being present in the next depth
+// scrape yet. m.mu must be held so _other deletions cannot race a concurrent
+// overflow Set for the same action.
+func (m *SQSMetrics) admitForThrottleGaugeBudgetLocked(queue string, action string) string {
+ m.throttleGaugeSeq++
+ m.throttleGaugeQueueSeq[queue] = m.throttleGaugeSeq
+ if m.throttleGaugeActionSeq[queue] == nil {
+ m.throttleGaugeActionSeq[queue] = map[string]uint64{}
+ }
+ m.throttleGaugeActionSeq[queue][action] = m.throttleGaugeSeq
+ if actions, ok := m.trackedThrottleGaugeQueues[queue]; ok {
+ actions[action] = struct{}{}
+ if m.removeThrottleOverflowActionLocked(queue, action) {
+ m.dropThrottleGaugeActionFor(sqsQueueOverflow, action)
+ }
+ return queue
+ }
+ if len(m.trackedThrottleGaugeQueues) >= sqsMaxTrackedQueues {
+ queues := m.overflowThrottleGaugeQueues[action]
+ if queues == nil {
+ queues = map[string]struct{}{}
+ m.overflowThrottleGaugeQueues[action] = queues
+ }
+ queues[queue] = struct{}{}
+ return sqsQueueOverflow
+ }
+ m.trackedThrottleGaugeQueues[queue] = map[string]struct{}{action: {}}
+ if m.removeThrottleOverflowActionLocked(queue, action) {
+ m.dropThrottleGaugeActionFor(sqsQueueOverflow, action)
+ }
+ return queue
+}
+
+// ThrottleGaugeSnapshotCutoff returns the current token-gauge observation
+// sequence. Config-change cleanup can use it to forget only gauges observed
+// before a reset point while preserving later request observations.
+func (m *SQSMetrics) ThrottleGaugeSnapshotCutoff() uint64 {
+ return m.throttleGaugeSnapshotCutoff()
+}
+
+func (m *SQSMetrics) throttleGaugeSnapshotCutoff() uint64 {
+ if m == nil {
+ return 0
+ }
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return m.throttleGaugeSeq
+}
+
+func (m *SQSMetrics) snapshotThrottleGaugeQueues(cutoff uint64) []string {
+ if m == nil {
+ return nil
+ }
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ seen := make(map[string]struct{}, len(m.trackedThrottleGaugeQueues))
+ queues := make([]string, 0, len(m.trackedThrottleGaugeQueues)+len(m.overflowThrottleGaugeQueues))
+ for queue := range m.trackedThrottleGaugeQueues {
+ if m.throttleGaugeQueueSeq[queue] > cutoff {
+ continue
+ }
+ seen[queue] = struct{}{}
+ queues = append(queues, queue)
+ }
+ for _, overflowQueues := range m.overflowThrottleGaugeQueues {
+ for queue := range overflowQueues {
+ if m.throttleGaugeQueueSeq[queue] > cutoff {
+ continue
+ }
+ if _, ok := seen[queue]; ok {
+ continue
+ }
+ seen[queue] = struct{}{}
+ queues = append(queues, queue)
+ }
+ }
+ return queues
+}
+
// dropGaugeStatesFor removes the three state-labelled series for
// label (a real queue name or sqsQueueOverflow) from the depth
// gauge. Single-line caller-readability wrapper — prometheus
@@ -371,6 +793,16 @@ func (m *SQSMetrics) dropGaugeStatesFor(label string) {
m.queueDepth.DeleteLabelValues(label, sqsQueueStateDelayed)
}
+func (m *SQSMetrics) dropThrottleGaugeActionsFor(label string) {
+ for _, action := range []string{SQSThrottleActionSend, SQSThrottleActionReceive, SQSThrottleActionDefault} {
+ m.dropThrottleGaugeActionFor(label, action)
+ }
+}
+
+func (m *SQSMetrics) dropThrottleGaugeActionFor(label string, action string) {
+ m.throttleTokens.DeleteLabelValues(label, action)
+}
+
// sqsValidPartitionAction returns true iff action is one of the
// stable label values. Keeps a typo at the call site (e.g.
// "Send" vs "send") from polluting the metric.
@@ -382,6 +814,14 @@ func sqsValidPartitionAction(action string) bool {
return false
}
+func sqsValidThrottleAction(action string) bool {
+ switch action {
+ case SQSThrottleActionSend, SQSThrottleActionReceive, SQSThrottleActionDefault:
+ return true
+ }
+ return false
+}
+
// sqsDepthObserveInterval is the default tick cadence for
// SQSObserver. 30 s mirrors sqsReaperInterval — dashboards are
// rarely refreshed faster, and tighter ticks just add catalog-scan
@@ -398,14 +838,16 @@ const sqsDepthObserveInterval = 30 * time.Second
type SQSObserver struct {
metrics *SQSMetrics
- mu sync.Mutex
- lastSeen map[string]struct{}
+ mu sync.Mutex
+ lastSeen map[string]struct{}
+ depthSeenThrottleQueues map[string]struct{}
}
func newSQSObserver(metrics *SQSMetrics) *SQSObserver {
return &SQSObserver{
- metrics: metrics,
- lastSeen: map[string]struct{}{},
+ metrics: metrics,
+ lastSeen: map[string]struct{}{},
+ depthSeenThrottleQueues: map[string]struct{}{},
}
}
@@ -456,6 +898,7 @@ func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) {
if o == nil || o.metrics == nil || source == nil {
return
}
+ throttleCutoff := o.metrics.throttleGaugeSnapshotCutoff()
snaps, ok := source.SnapshotQueueDepths(ctx)
if !ok {
// Source signalled "skip this tick" (transient catalog-scan
@@ -479,12 +922,9 @@ func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) {
// at least one interval even when capacity opens up the same
// tick. Reclaiming first lets phase 2's admissions reuse the
// freed slots immediately.
+ sourceInactive := snaps == nil
o.mu.Lock()
- for prev := range o.lastSeen {
- if _, ok := current[prev]; !ok {
- o.metrics.ForgetQueue(prev)
- }
- }
+ o.forgetMissingQueuesLocked(current, throttleCutoff, sourceInactive)
o.lastSeen = current
o.mu.Unlock()
// Phase 2: emit gauges for the current tick. Slots freed in
@@ -496,3 +936,69 @@ func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) {
o.metrics.ObserveQueueDepth(snap.Queue, snap.Visible, snap.NotVisible, snap.Delayed)
}
}
+
+func (o *SQSObserver) forgetMissingQueuesLocked(current map[string]struct{}, throttleCutoff uint64, sourceInactive bool) {
+ if sourceInactive {
+ throttleQueues, _ := o.snapshotThrottleQueues(^uint64(0))
+ o.forgetDepthQueuesMissingFrom(current, false)
+ o.forgetInactiveSourceThrottleQueues(current, throttleQueues)
+ clear(o.depthSeenThrottleQueues)
+ return
+ }
+ throttleQueues, _ := o.snapshotThrottleQueues(throttleCutoff)
+ _, currentThrottleQueues := o.snapshotThrottleQueues(^uint64(0))
+ o.forgetDepthSeenThrottleQueuesWithoutGauge(currentThrottleQueues)
+ o.forgetDepthQueuesMissingFrom(current, true)
+ o.forgetThrottleOnlyQueues(current, throttleQueues)
+}
+
+func (o *SQSObserver) snapshotThrottleQueues(throttleCutoff uint64) ([]string, map[string]struct{}) {
+ throttleQueues := o.metrics.snapshotThrottleGaugeQueues(throttleCutoff)
+ activeThrottleQueues := make(map[string]struct{}, len(throttleQueues))
+ for _, queue := range throttleQueues {
+ activeThrottleQueues[queue] = struct{}{}
+ }
+ return throttleQueues, activeThrottleQueues
+}
+
+func (o *SQSObserver) forgetDepthSeenThrottleQueuesWithoutGauge(activeThrottleQueues map[string]struct{}) {
+ for queue := range o.depthSeenThrottleQueues {
+ if _, ok := activeThrottleQueues[queue]; !ok {
+ delete(o.depthSeenThrottleQueues, queue)
+ }
+ }
+}
+
+func (o *SQSObserver) forgetDepthQueuesMissingFrom(current map[string]struct{}, preserveThrottle bool) {
+ for prev := range o.lastSeen {
+ if _, ok := current[prev]; !ok {
+ if preserveThrottle {
+ o.depthSeenThrottleQueues[prev] = struct{}{}
+ }
+ o.metrics.ForgetQueue(prev)
+ }
+ }
+}
+
+func (o *SQSObserver) forgetInactiveSourceThrottleQueues(current map[string]struct{}, throttleQueues []string) {
+ for _, prev := range throttleQueues {
+ if _, ok := current[prev]; ok {
+ continue
+ }
+ o.metrics.ForgetThrottleQueue(prev)
+ }
+}
+
+func (o *SQSObserver) forgetThrottleOnlyQueues(current map[string]struct{}, throttleQueues []string) {
+ for _, prev := range throttleQueues {
+ if _, ok := current[prev]; !ok {
+ if _, wasDepthSeen := o.lastSeen[prev]; wasDepthSeen {
+ continue
+ }
+ if _, wasDepthSeen := o.depthSeenThrottleQueues[prev]; wasDepthSeen {
+ continue
+ }
+ o.metrics.ForgetThrottleQueue(prev)
+ }
+ }
+}
diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go
index 65be82202..c4e13c915 100644
--- a/monitoring/sqs_test.go
+++ b/monitoring/sqs_test.go
@@ -9,9 +9,48 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/testutil"
+ dto "github.com/prometheus/client_model/go"
"github.com/stretchr/testify/require"
)
+func gatheredThrottleTokenValue(t *testing.T, reg *prometheus.Registry, labels map[string]string) (float64, bool) {
+ t.Helper()
+ mfs, err := reg.Gather()
+ require.NoError(t, err)
+ for _, mf := range mfs {
+ if mf.GetName() != "elastickv_sqs_throttle_tokens_remaining" {
+ continue
+ }
+ for _, metric := range mf.GetMetric() {
+ if !metricLabelsMatch(metric.GetLabel(), labels) {
+ continue
+ }
+ switch {
+ case metric.GetGauge() != nil:
+ return metric.GetGauge().GetValue(), true
+ case metric.GetCounter() != nil:
+ return metric.GetCounter().GetValue(), true
+ default:
+ return 0, true
+ }
+ }
+ }
+ return 0, false
+}
+
+func metricLabelsMatch(labels []*dto.LabelPair, want map[string]string) bool {
+ if len(labels) != len(want) {
+ return false
+ }
+ for _, label := range labels {
+ got, ok := want[label.GetName()]
+ if !ok || got != label.GetValue() {
+ return false
+ }
+ }
+ return true
+}
+
// TestSQSMetrics_ObservePartitionMessage_IncrementsByLabelTriple
// pins the basic counter contract: each (queue, partition,
// action) tuple gets its own series and only the matching
@@ -83,6 +122,7 @@ func TestSQSMetrics_NilReceiverIsSafe(t *testing.T) {
var m *SQSMetrics
require.NotPanics(t, func() {
m.ObservePartitionMessage("q.fifo", 0, SQSPartitionActionSend)
+ m.ObserveThrottleDecision("q.fifo", SQSThrottleActionSend, 1, true)
})
}
@@ -127,19 +167,303 @@ func TestSQSMetrics_RegistryWiring(t *testing.T) {
require.NotNil(t, obs)
obs.ObservePartitionMessage("q.fifo", 3, SQSPartitionActionReceive)
+ throttleObs, ok := obs.(SQSThrottleObserver)
+ require.True(t, ok)
+ throttleObs.ObserveThrottleDecision("q.fifo", SQSThrottleActionSend, 4, true)
mfs, err := r.Gatherer().Gather()
require.NoError(t, err)
- var found bool
+ var foundPartition bool
+ var foundThrottleCounter bool
+ var foundThrottleGauge bool
for _, mf := range mfs {
- if !strings.HasPrefix(mf.GetName(), "elastickv_sqs_partition_messages_total") {
+ switch {
+ case strings.HasPrefix(mf.GetName(), "elastickv_sqs_partition_messages_total"):
+ foundPartition = true
+ require.Len(t, mf.GetMetric(), 1)
+ case strings.HasPrefix(mf.GetName(), "elastickv_sqs_throttled_requests_total"):
+ foundThrottleCounter = true
+ require.Len(t, mf.GetMetric(), 1)
+ case strings.HasPrefix(mf.GetName(), "elastickv_sqs_throttle_tokens_remaining"):
+ foundThrottleGauge = true
+ require.Len(t, mf.GetMetric(), 1)
+ default:
continue
}
- found = true
- require.Len(t, mf.GetMetric(), 1)
}
- require.True(t, found,
+ require.True(t, foundPartition,
"elastickv_sqs_partition_messages_total must be registered on the public Registry")
+ require.True(t, foundThrottleCounter,
+ "elastickv_sqs_throttled_requests_total must be registered on the public Registry")
+ require.True(t, foundThrottleGauge,
+ "elastickv_sqs_throttle_tokens_remaining must be registered on the public Registry")
+}
+
+func TestSQSMetrics_ObserveThrottleDecision_EmitsCounterAndGauge(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 9, false)
+ require.InDelta(t, 9.0, testutil.ToFloat64(m.throttleTokens.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001)
+ require.InDelta(t, 0.0, testutil.ToFloat64(m.throttledRequests.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 0, true)
+ require.InDelta(t, 0.0, testutil.ToFloat64(m.throttleTokens.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001)
+ require.InDelta(t, 1.0, testutil.ToFloat64(m.throttledRequests.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001)
+}
+
+func TestSQSMetrics_ObserveThrottleRejection_EmitsCounterOnly(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleRejection("orders.fifo", SQSThrottleActionSend)
+
+ require.InDelta(t, 1.0, testutil.ToFloat64(m.throttledRequests.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001)
+ count, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, count, "stale rejection accounting must not recreate a token gauge")
+}
+
+func TestSQSMetrics_ObserveThrottleDecision_AllowedDoesNotConsumeCounterBudget(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ for i := 0; i < sqsMaxTrackedQueues; i++ {
+ m.ObserveThrottleDecision("allow-"+strconv.Itoa(i)+".fifo", SQSThrottleActionSend, 1, false)
+ }
+ require.Empty(t, m.trackedCounterQueues, "allowed-only throttle decisions must not consume counter label budget")
+
+ m.ObserveThrottleDecision("blocked.fifo", SQSThrottleActionSend, 0, true)
+ require.Empty(t, m.trackedCounterQueues, "throttle rejects must not consume the partition counter label budget")
+ require.Len(t, m.trackedThrottleCounterQueues, 1, "throttle rejects consume the throttle counter label budget")
+ require.InDelta(t, 1.0,
+ testutil.ToFloat64(m.throttledRequests.WithLabelValues("blocked.fifo", SQSThrottleActionSend)),
+ 0.001,
+ "first actual reject must keep its real queue label instead of overflowing after allowed-only decisions")
+}
+
+func TestSQSMetrics_ThrottleCounterBudgetIsIndependent(t *testing.T) {
+ t.Parallel()
+
+ t.Run("throttle rejects do not hide later partition counters", func(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ for i := 0; i < sqsMaxTrackedQueues; i++ {
+ m.ObserveThrottleDecision("throttle-"+strconv.Itoa(i)+".fifo", SQSThrottleActionSend, 0, true)
+ }
+ require.Len(t, m.trackedThrottleCounterQueues, sqsMaxTrackedQueues)
+ require.Empty(t, m.trackedCounterQueues)
+
+ m.ObservePartitionMessage("partition.fifo", 0, SQSPartitionActionSend)
+ require.InDelta(t, 1.0,
+ testutil.ToFloat64(m.partitionMessages.WithLabelValues("partition.fifo", "0", SQSPartitionActionSend)),
+ 0.001,
+ "partition counter must still get a real queue label after throttle counter budget fills")
+ })
+
+ t.Run("partition counters do not hide later throttle rejects", func(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ for i := 0; i < sqsMaxTrackedQueues; i++ {
+ m.ObservePartitionMessage("partition-"+strconv.Itoa(i)+".fifo", 0, SQSPartitionActionSend)
+ }
+ require.Len(t, m.trackedCounterQueues, sqsMaxTrackedQueues)
+ require.Empty(t, m.trackedThrottleCounterQueues)
+
+ m.ObserveThrottleDecision("blocked.fifo", SQSThrottleActionSend, 0, true)
+ require.InDelta(t, 1.0,
+ testutil.ToFloat64(m.throttledRequests.WithLabelValues("blocked.fifo", SQSThrottleActionSend)),
+ 0.001,
+ "throttle counter must still get a real queue label after partition counter budget fills")
+ })
+}
+
+func TestSQSMetrics_ObserveThrottleDecision_DropsInvalidLabels(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleDecision("", SQSThrottleActionSend, 1, true)
+ m.ObserveThrottleDecision("orders.fifo", "Send", 1, true)
+
+ counterCount, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttled_requests_total")
+ require.NoError(t, err)
+ require.Equal(t, 0, counterCount)
+ gaugeCount, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, gaugeCount)
+}
+
+func TestSQSMetrics_ForgetThrottleAction_DropsGaugeAndFreesSlot(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 5, false)
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionReceive, 7, false)
+ require.Len(t, m.trackedThrottleGaugeQueues, 1)
+
+ m.ForgetThrottleAction("orders.fifo", SQSThrottleActionSend)
+ _, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.False(t, found, "disabled send bucket must drop the stale token gauge")
+ remaining, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionReceive,
+ })
+ require.True(t, found, "other configured actions for the same queue must survive")
+ require.InDelta(t, 7.0, remaining, 0.001)
+ require.Len(t, m.trackedThrottleGaugeQueues, 1, "queue stays admitted while another action remains")
+
+ m.ForgetThrottleAction("orders.fifo", SQSThrottleActionReceive)
+ count, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, count)
+ require.Empty(t, m.trackedThrottleGaugeQueues, "last disabled action frees the throttle gauge queue slot")
+}
+
+func TestSQSMetrics_ForgetThrottleActionBefore_PreservesFreshActionGauge(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 5, false)
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionReceive, 7, false)
+ cutoff := m.ThrottleGaugeSnapshotCutoff()
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 3, false)
+
+ m.ForgetThrottleActionBefore("orders.fifo", SQSThrottleActionSend, cutoff)
+ m.ForgetThrottleActionBefore("orders.fifo", SQSThrottleActionReceive, cutoff)
+
+ send, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "post-cutoff send gauge must survive reset cleanup")
+ require.InDelta(t, 3.0, send, 0.001)
+ _, found = gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionReceive,
+ })
+ require.False(t, found, "pre-cutoff receive gauge should still be reset")
+}
+
+func TestSQSMetrics_SyncThrottleActions_DropsDisabledWithoutTraffic(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 5, false)
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionReceive, 7, false)
+
+ m.SyncThrottleActions("orders.fifo", []string{SQSThrottleActionSend})
+ send, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "configured send bucket must keep its token gauge")
+ require.InDelta(t, 5.0, send, 0.001)
+ _, found = gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionReceive,
+ })
+ require.False(t, found, "disabled receive bucket must be removed without waiting for more traffic")
+ require.Len(t, m.trackedThrottleGaugeQueues, 1, "queue stays admitted while send remains configured")
+
+ m.SyncThrottleActions("orders.fifo", nil)
+ count, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, count)
+ require.Empty(t, m.trackedThrottleGaugeQueues, "disabling the last action frees the gauge slot")
+}
+
+func TestSQSMetrics_ForgetThrottleQueue_OverflowThrottleGaugeClearsPerAction(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ for i := 0; i < sqsMaxTrackedQueues; i++ {
+ m.ObserveThrottleDecision("real-"+strconv.Itoa(i)+".fifo", SQSThrottleActionSend, 1, false)
+ }
+ require.Len(t, m.trackedThrottleGaugeQueues, sqsMaxTrackedQueues)
+
+ m.ObserveThrottleDecision("overflow-a.fifo", SQSThrottleActionSend, 10, false)
+ m.ObserveThrottleDecision("overflow-b.fifo", SQSThrottleActionReceive, 20, false)
+
+ send, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found)
+ require.InDelta(t, 10.0, send, 0.001)
+ receive, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionReceive,
+ })
+ require.True(t, found)
+ require.InDelta(t, 20.0, receive, 0.001)
+
+ m.ForgetThrottleQueue("overflow-a.fifo")
+ _, found = gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionSend,
+ })
+ require.False(t, found, "last overflow send queue disappearing must drop only _other/send")
+ receive, found = gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionReceive,
+ })
+ require.True(t, found, "_other/receive must survive while another overflow receive queue remains")
+ require.InDelta(t, 20.0, receive, 0.001)
+
+ m.ForgetThrottleQueue("overflow-b.fifo")
+ _, found = gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionReceive,
+ })
+ require.False(t, found, "last overflow receive queue disappearing must drop _other/receive")
+}
+
+func TestSQSMetrics_ThrottleOverflowPromotionPreservesOtherActionGauge(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ for i := 0; i < sqsMaxTrackedQueues; i++ {
+ m.ObserveThrottleDecision("real-"+strconv.Itoa(i)+".fifo", SQSThrottleActionSend, 1, false)
+ }
+ m.ObserveThrottleDecision("overflow.fifo", SQSThrottleActionSend, 10, false)
+ m.ObserveThrottleDecision("overflow.fifo", SQSThrottleActionReceive, 20, false)
+
+ m.ForgetThrottleQueue("real-0.fifo")
+ m.ObserveThrottleDecision("overflow.fifo", SQSThrottleActionSend, 8, false)
+
+ send, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "overflow.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "promoted action should move to the real queue label")
+ require.InDelta(t, 8.0, send, 0.001)
+ _, found = gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionSend,
+ })
+ require.False(t, found, "the promoted send overflow gauge should be removed")
+ receive, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": sqsQueueOverflow,
+ "action": SQSThrottleActionReceive,
+ })
+ require.True(t, found, "unrelated receive overflow gauge must survive until receive traffic is promoted")
+ require.InDelta(t, 20.0, receive, 0.001)
}
// TestSQSMetrics_ObserveQueueDepth_EmitsThreeStates pins that one
@@ -190,19 +514,20 @@ func TestSQSMetrics_ObserveQueueDepth_DropsEmptyQueue(t *testing.T) {
}
// TestSQSMetrics_ForgetQueue_DropsThreeSeries pins that ForgetQueue
-// (a) removes all three state-labelled series so a deleted queue
-// stops reporting a frozen backlog, (b) frees the cardinality-
-// budget slot so a churn-heavy deployment doesn't permanently
-// exhaust the 512-entry budget, and (c) leaves the
-// (queue, partition, action) counter alone (cumulative-by-design).
+// (a) removes all three state-labelled depth series so a deleted queue stops
+// reporting a frozen backlog, (b) frees the depth cardinality-budget slot so a
+// churn-heavy deployment doesn't permanently exhaust the 512-entry budget, and
+// (c) leaves cumulative counters and throttle-token gauges alone.
func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) {
t.Parallel()
reg := prometheus.NewRegistry()
m := newSQSMetrics(reg)
m.ObserveQueueDepth("orders.fifo", 3, 0, 0)
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 4, true)
m.ObservePartitionMessage("orders.fifo", 0, SQSPartitionActionSend)
require.Len(t, m.trackedDepthQueues, 1, "queue must have been admitted to the depth budget on ObserveQueueDepth")
+ require.Len(t, m.trackedThrottleGaugeQueues, 1, "queue must have been admitted to the throttle gauge budget on ObserveThrottleDecision")
require.Len(t, m.trackedCounterQueues, 1, "queue must have been admitted to the counter budget on ObservePartitionMessage")
m.ForgetQueue("orders.fifo")
@@ -210,10 +535,16 @@ func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) {
count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages")
require.NoError(t, err)
require.Equal(t, 0, count, "ForgetQueue must drop every state series for the queue")
+ throttleGaugeCount, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 1, throttleGaugeCount, "ForgetQueue must not drop throttle-token gauges for a depth-only miss")
require.Empty(t, m.trackedDepthQueues, "ForgetQueue must free the depth-side cardinality slot")
+ require.Len(t, m.trackedThrottleGaugeQueues, 1, "ForgetQueue must not free the throttle-gauge cardinality slot")
require.Len(t, m.trackedCounterQueues, 1, "ForgetQueue must NOT free the counter-side slot (counters are cumulative)")
require.InDelta(t, 1.0, testutil.ToFloat64(m.partitionMessages.WithLabelValues("orders.fifo", "0", SQSPartitionActionSend)), 0.001,
"counter must survive ForgetQueue (cumulative metric)")
+ require.InDelta(t, 1.0, testutil.ToFloat64(m.throttledRequests.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001,
+ "throttled-request counter must survive ForgetQueue (cumulative metric)")
// Observe again post-forget: the new series must be keyed on
// the real name (slot was freed), NOT on the _other overflow
@@ -224,6 +555,25 @@ func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) {
"post-forget Observe must re-emit under the real queue name")
}
+func TestSQSMetrics_ForgetThrottleQueue_DropsTokenGauges(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 4, true)
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionReceive, 7, false)
+ require.Len(t, m.trackedThrottleGaugeQueues, 1)
+
+ m.ForgetThrottleQueue("orders.fifo")
+
+ count, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, count, "ForgetThrottleQueue must drop every token gauge for the queue")
+ require.Empty(t, m.trackedThrottleGaugeQueues, "ForgetThrottleQueue must free the throttle-gauge cardinality slot")
+ require.InDelta(t, 1.0, testutil.ToFloat64(m.throttledRequests.WithLabelValues("orders.fifo", SQSThrottleActionSend)), 0.001,
+ "throttled-request counter must survive ForgetThrottleQueue (cumulative metric)")
+}
+
// TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp pins the
// overflow-collision safety: a queue that hit the cardinality cap
// and got collapsed onto the _other label has no individual series
@@ -532,6 +882,14 @@ func (f *fakeDepthSource) SnapshotQueueDepths(_ context.Context) ([]SQSQueueDept
return t.snaps, t.ok
}
+type callbackDepthSource struct {
+ fn func() ([]SQSQueueDepth, bool)
+}
+
+func (f callbackDepthSource) SnapshotQueueDepths(context.Context) ([]SQSQueueDepth, bool) {
+ return f.fn()
+}
+
// TestSQSObserver_ObserveOnce_EmitsAndForgets pins the observer's
// state machine: queues present in the current tick get gauges
// emitted, queues that disappeared since the last tick get
@@ -598,11 +956,146 @@ func TestSQSObserver_ObserveOnce_LeaderStepDownClearsAll(t *testing.T) {
count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages")
require.NoError(t, err)
require.Equal(t, 6, count, "tick 1: 2 queues × 3 states = 6 series")
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 4, false)
obs.ObserveOnce(source)
count, err = testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages")
require.NoError(t, err)
require.Equal(t, 0, count, "tick 2 (leader step-down): all gauges cleared")
+ throttleCount, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, throttleCount, "tick 2 (leader step-down): throttle gauges cleared with depth gauges")
+}
+
+func TestSQSObserver_ObserveOnce_ForgetsThrottleOnlyQueues(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+ obs := newSQSObserver(m)
+
+ m.ObserveThrottleDecision("short-lived.fifo", SQSThrottleActionSend, 4, false)
+ count, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 1, count, "sanity: throttle-only queue emitted a token gauge before the depth tick")
+
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{}, ok: true}},
+ })
+
+ count, err = testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, count, "successful depth tick must clear throttle-only queues absent from the snapshot")
+ require.Empty(t, m.trackedThrottleGaugeQueues, "throttle gauge budget slot must be reclaimed for the short-lived queue")
+}
+
+func TestSQSObserver_ObserveOnce_PreservesThrottleGaugeNewerThanDepthSnapshot(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+ obs := newSQSObserver(m)
+
+ obs.ObserveOnce(callbackDepthSource{fn: func() ([]SQSQueueDepth, bool) {
+ m.ObserveThrottleDecision("newer.fifo", SQSThrottleActionSend, 4, false)
+ return []SQSQueueDepth{}, true
+ }})
+
+ value, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "newer.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "a throttle gauge emitted after the cleanup cutoff must survive until the next depth tick")
+ require.InDelta(t, 4.0, value, 0.001)
+
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{}, ok: true}},
+ })
+ count, err := testutil.GatherAndCount(reg, "elastickv_sqs_throttle_tokens_remaining")
+ require.NoError(t, err)
+ require.Equal(t, 0, count, "the next successful depth tick can clear the now-old throttle-only gauge")
+}
+
+func TestSQSObserver_ObserveOnce_DepthMissPreservesThrottleGauge(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+ obs := newSQSObserver(m)
+
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 4, false)
+ source := &fakeDepthSource{
+ ticks: []fakeDepthTick{
+ {snaps: []SQSQueueDepth{{Queue: "orders.fifo", Visible: 1}}, ok: true},
+ {snaps: []SQSQueueDepth{}, ok: true},
+ {snaps: []SQSQueueDepth{}, ok: true},
+ },
+ }
+
+ obs.ObserveOnce(source)
+ obs.ObserveOnce(source)
+ obs.ObserveOnce(source)
+
+ depthCount, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages")
+ require.NoError(t, err)
+ require.Equal(t, 0, depthCount, "depth miss still clears depth gauges through ForgetQueue")
+ value, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "depth-only miss must not delete throttle-token gauges")
+ require.InDelta(t, 4.0, value, 0.001)
+ require.Len(t, m.trackedThrottleGaugeQueues, 1, "depth-only miss must not reclaim the throttle-gauge budget slot")
+}
+
+func TestSQSObserver_ObserveOnce_DepthSeenFreshGaugeAfterCutoffSurvivesMiss(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+ obs := newSQSObserver(m)
+
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{{Queue: "orders.fifo", Visible: 1}}, ok: true}},
+ })
+ obs.ObserveOnce(callbackDepthSource{fn: func() ([]SQSQueueDepth, bool) {
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 6, false)
+ return []SQSQueueDepth{}, true
+ }})
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{}, ok: true}},
+ })
+
+ value, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "fresh token gauge for a previously depth-seen queue must survive repeated depth misses")
+ require.InDelta(t, 6.0, value, 0.001)
+}
+
+func TestSQSObserver_ObserveOnce_DepthSeenMarkerSurvivesFreshGaugeDuringLaterMiss(t *testing.T) {
+ t.Parallel()
+ reg := prometheus.NewRegistry()
+ m := newSQSMetrics(reg)
+ obs := newSQSObserver(m)
+
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{{Queue: "orders.fifo", Visible: 1}}, ok: true}},
+ })
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{}, ok: true}},
+ })
+ obs.ObserveOnce(callbackDepthSource{fn: func() ([]SQSQueueDepth, bool) {
+ m.ObserveThrottleDecision("orders.fifo", SQSThrottleActionSend, 7, false)
+ return []SQSQueueDepth{}, true
+ }})
+ obs.ObserveOnce(&fakeDepthSource{
+ ticks: []fakeDepthTick{{snaps: []SQSQueueDepth{}, ok: true}},
+ })
+
+ value, found := gatheredThrottleTokenValue(t, reg, map[string]string{
+ "queue": "orders.fifo",
+ "action": SQSThrottleActionSend,
+ })
+ require.True(t, found, "fresh token gauge for a previously depth-seen queue must keep the depth-seen marker")
+ require.InDelta(t, 7.0, value, 0.001)
}
// TestSQSObserver_ObserveOnce_TransientScanErrorPreservesGauges