docs(sqs): proposals for Phase 3.C (throttling) + 3.D (split-queue FIFO) (#664)

## Summary

**Docs-only PR.** Two design proposals for the remaining Phase 3 SQS
items per
[`docs/design/2026_04_24_proposed_sqs_compatible_adapter.md`](https://github.com/bootjp/elastickv/blob/main/docs/design/2026_04_24_proposed_sqs_compatible_adapter.md)
Section 14 (Phase 3 bullets). Both items were explicitly called out as
needing separate design docs before any implementation work; this PR
lands those proposals so the implementation PRs have a reviewed
architecture to build on.

### 3.C — Per-queue throttling and tenant fairness
([proposal](docs/design/2026_04_26_proposed_sqs_per_queue_throttling.md))

Per-queue token-bucket throttling configured on queue meta (no separate
keyspace), evaluated at the SQS adapter layer on the leader (no Raft per
request), surfaced as the AWS `Throttling` error envelope so SDK
retry/backoff just works.

Key decisions:
- **Default-off**. Existing queues are unaffected; operators opt in per
queue via `SetQueueAttributes`.
- **Per-action buckets** (Send / Receive / Default) so a slow consumer
cannot pin the producer.
- **Per-leader buckets, no replication**. Worst case on failover: one
extra burst on the new leader. Acceptable per AWS-equivalent behaviour
at region failover boundaries; replicating would cost a Raft commit per
`SendMessage`.
- **Batch verbs charge by entry count**, not call count, with
all-or-nothing rejection (matches AWS).
- **Admin-only configuration plane**. Standard SQS clients see
`InvalidAttributeName` on the `Throttle*` attributes (matches AWS
behaviour for unknown attributes); the data-plane enforcement runs for
everyone.

### 3.D — Split-queue FIFO (high-throughput FIFO)
([proposal](docs/design/2026_04_26_proposed_sqs_split_queue_fifo.md))

Per-`MessageGroupId` hash partitioning across multiple Raft groups,
mirroring AWS High Throughput FIFO. Within-group ordering preserved;
across-group throughput scales with the partition count.

Key decisions:
- **Existing single-partition FIFO queues stay byte-identical**
(`PartitionCount = 0` path is the legacy layout; no migration runs
implicitly).
- **Power-of-two partition counts only** (1, 2, 4, 8, 16, 32) so the
routing step is `hash & (N-1)` and future offline rebuilds stay
tractable.
- **Partition count is immutable after first SendMessage**. Live
re-partitioning would break ordering for in-flight messages of every
group whose hash bucket changed; out of scope.
- **Multi-PR rollout plan** with an explicit "gate of no return" called
out at PR 5 (the data-plane PR). PRs 1–4 are reversible no-ops on data
layout; once a partitioned FIFO holds real data, rollback means draining
and recreating the queue.
- **FNV-1a hash** (deterministic across processes / Go versions /
architectures). Risk of attacker-controlled `MessageGroupId` pinning all
traffic to one partition is documented and accepted (the feature is for
cooperative operators).

## Test plan

- [x] Markdown renders correctly on GitHub (manually previewed).
- [x] Cross-references resolve (the partial-doc rename in PR #659 is in
flight; both proposals reference the partial filename — they will
resolve once #659 merges, otherwise resolve to the current `_proposed_`
filename for a 1-character mismatch that is fine to leave for now).
- [x] No code changes; CI is irrelevant beyond the markdown lint pass.

## Self-review

This is a docs-only PR; the 5-lens self-review collapses to:

1. **Data loss / Concurrency / Performance / Consistency** — N/A, no
code touched.
2. **Test coverage** — N/A, no code touched. The proposals themselves
include a Testing Strategy section (§6 / §9) so the implementation PRs
have explicit acceptance criteria.

## Stacking

This PR is **independent** of #650, #659, and #662. Branched from
current `main`. Merge whenever ready — landing the proposal docs early
lets reviewers comment on the architecture before the implementation PRs
go up.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Documentation**
* Added detailed designs for optional per-queue token-bucket throttling
(rules, batch semantics, AWS-shaped Throttling errors with Retry-After,
metrics, rollout default-off) and for HT‑FIFO “split-queue” partitioning
(routing, immutability, gating/rollout).
* **New Features**
* Optional queue-level send/receive throttling with batch-aware charging
and Retry-After; HT‑FIFO partitioning with deterministic partition
routing and immutable partition attributes.
* **Tests**
* Extensive unit and integration tests for throttling, HT‑FIFO
partitioning, attribute validation, immutability, idempotency, and cache
invalidation.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

adapter/sqs.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/bootjp/elastickv/kv"
@@ -55,6 +56,11 @@ const (
 	sqsErrInternalFailure    = "InternalFailure"
 	sqsErrServiceUnavailable = "ServiceUnavailable"
 	sqsErrMalformedRequest   = "MalformedQueryString"
+	// sqsErrThrottling is the per-queue rate-limit rejection code.
+	// Returned with HTTP 400 and a Retry-After header derived from the
+	// bucket's refillRate + the request's charge count (see
+	// computeRetryAfter in sqs_throttle.go for the formula).
+	sqsErrThrottling = "Throttling"
 )
 
 type SQSServerOption func(*SQSServer)
@@ -76,6 +82,11 @@ type SQSServer struct {
 	// goroutines without ordering between them.
 	reaperCtx    context.Context
 	reaperCancel context.CancelFunc
+	// throttle is the per-queue rate-limit bucket store. Always
+	// non-nil; charge() short-circuits when the queue's meta has no
+	// throttle config so unconfigured queues pay one nil-check per
+	// request and nothing else (see sqs_throttle.go).
+	throttle *bucketStore
 }
 
 // WithSQSLeaderMap configures the Raft-address-to-SQS-address mapping used to
@@ -98,6 +109,7 @@ func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordin
 		coordinator:  coordinate,
 		reaperCtx:    reaperCtx,
 		reaperCancel: reaperCancel,
+		throttle:     newBucketStoreDefault(),
 	}
 	s.targetHandlers = map[string]func(http.ResponseWriter, *http.Request){
 		sqsCreateQueueTarget:               s.createQueue,
@@ -131,6 +143,10 @@ func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordin
 
 func (s *SQSServer) Run() error {
 	s.startReaper(s.reaperCtx)
+	// Throttle bucket idle-evict runs on a background ticker so the
+	// request hot path never pays the O(N) sweep cost. Cleaned up by
+	// the same reaperCtx cancellation that stops the message reaper.
+	go s.throttle.runSweepLoop(s.reaperCtx)
 	if err := s.httpServer.Serve(s.listen); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		return errors.WithStack(err)
 	}
@@ -302,3 +318,19 @@ func writeSQSError(w http.ResponseWriter, status int, code string, message strin
 	w.WriteHeader(status)
 	_ = json.NewEncoder(w).Encode(resp)
 }
+
+// writeSQSThrottlingError emits the rate-limit rejection envelope: 400
+// + the AWS-shaped JSON error body + a Retry-After header carrying
+// the integer-second wait derived from the bucket's refill rate and
+// the request's charge count. The action argument is the bucket-action
+// vocabulary ("Send" | "Receive" | "*") so the operator-visible
+// message names the bucket that ran out, not just the queue.
+func writeSQSThrottlingError(w http.ResponseWriter, queue, action string, retryAfter time.Duration) {
+	if retryAfter < time.Second {
+		retryAfter = time.Second
+	}
+	secs := int(retryAfter / time.Second)
+	w.Header().Set("Retry-After", strconv.Itoa(secs))
+	writeSQSError(w, http.StatusBadRequest, sqsErrThrottling,
+		"Rate exceeded for queue '"+queue+"' action '"+action+"'")
+}

adapter/sqs_catalog.go

@@ -136,6 +136,29 @@ func TestSQSServer_CatalogCreateIsIdempotent(t *testing.T) {
 	if got, _ := out3["__type"].(string); got != sqsErrQueueNameExists {
 		t.Fatalf("differing-attrs error type: got %q want %q", got, sqsErrQueueNameExists)
 	}
+
+	// Fourth call: same name, same non-throttle attrs as the original
+	// create, but different Throttle* values. The original create had
+	// no Throttle config; this one adds one. throttleConfigEqual must
+	// notice the diff and the call must reject as QueueNameExists.
+	// Without this case a bug in throttleConfigEqual (e.g. always
+	// returning true) would slip past the existing VisibilityTimeout-
+	// only test.
+	withThrottle := map[string]any{
+		"QueueName": "idempotent",
+		"Attributes": map[string]string{
+			"VisibilityTimeout":           "60",
+			"ThrottleSendCapacity":        "10",
+			"ThrottleSendRefillPerSecond": "1",
+		},
+	}
+	status4, out4 := callSQS(t, node, sqsCreateQueueTarget, withThrottle)
+	if status4 != http.StatusBadRequest {
+		t.Fatalf("re-create with added Throttle*: got %d want 400; body %v", status4, out4)
+	}
+	if got, _ := out4["__type"].(string); got != sqsErrQueueNameExists {
+		t.Fatalf("Throttle*-diff error type: got %q want %q", got, sqsErrQueueNameExists)
+	}
 }
 
 func TestSQSServer_CatalogGetAndSetAttributes(t *testing.T) {

adapter/sqs_catalog_test.go

@@ -294,33 +294,67 @@ type sqsChangeVisibilityInput struct {
 
 // ------------------------ handlers ------------------------
 
-func (s *SQSServer) sendMessage(w http.ResponseWriter, r *http.Request) {
+// prepareSendMessage decodes the SendMessage payload and resolves
+// the queue name. Throttle charging happens after the meta load in
+// validateSend so we don't pay an extra meta read just to discover
+// throttling is off.
+func (s *SQSServer) prepareSendMessage(w http.ResponseWriter, r *http.Request) (sqsSendMessageInput, string, bool) {
 	var in sqsSendMessageInput
 	if err := decodeSQSJSONInput(r, &in); err != nil {
 		writeSQSErrorFromErr(w, err)
-		return
+		return in, "", false
 	}
 	queueName, err := queueNameFromURL(in.QueueUrl)
 	if err != nil {
 		writeSQSErrorFromErr(w, err)
-		return
+		return in, "", false
 	}
+	return in, queueName, true
+}
+
+// validateSend loads queue meta, runs the throttle charge against
+// the loaded throttle config (no extra meta read), then validates
+// message attributes / FIFO params and resolves the delay. Returns
+// ok=false if any step has already written the error response.
+//
+// Throttle check sits AFTER the meta load (so we have the throttle
+// config) and AFTER the QueueDoesNotExist branch (so a missing
+// queue is reported as 400 QueueDoesNotExist, not as a Throttling
+// 400 against a non-existent bucket). It still sits OUTSIDE the
+// 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) {
 	meta, readTS, apiErr := s.loadQueueMetaForSend(r.Context(), queueName, []byte(in.MessageBody))
 	if apiErr != nil {
 		writeSQSErrorFromErr(w, apiErr)
-		return
+		return nil, 0, 0, false
+	}
+	if !s.chargeQueueWithThrottle(w, queueName, bucketActionSend, 1, meta.Throttle, meta.Incarnation) {
+		return nil, 0, 0, false
 	}
 	if apiErr := validateMessageAttributes(in.MessageAttributes); apiErr != nil {
 		writeSQSErrorFromErr(w, apiErr)
-		return
+		return nil, 0, 0, false
 	}
 	if apiErr := validateSendFIFOParams(meta, in); apiErr != nil {
 		writeSQSErrorFromErr(w, apiErr)
-		return
+		return nil, 0, 0, false
 	}
 	delay, apiErr := resolveSendDelay(meta, in.DelaySeconds)
 	if apiErr != nil {
 		writeSQSErrorFromErr(w, apiErr)
+		return nil, 0, 0, false
+	}
+	return meta, readTS, delay, true
+}
+
+func (s *SQSServer) sendMessage(w http.ResponseWriter, r *http.Request) {
+	in, queueName, ok := s.prepareSendMessage(w, r)
+	if !ok {
+		return
+	}
+	meta, readTS, delay, ok := s.validateSend(w, r, queueName, in)
+	if !ok {
 		return
 	}
 	if meta.IsFIFO {
@@ -530,6 +564,13 @@ func (s *SQSServer) receiveMessage(w http.ResponseWriter, r *http.Request) {
 		writeSQSError(w, http.StatusBadRequest, sqsErrQueueDoesNotExist, "queue does not exist")
 		return
 	}
+	// Throttle check uses the loaded meta's throttle config so we
+	// 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) {
+		return
+	}
 	max, maxErr := resolveReceiveMaxMessages(in.MaxNumberOfMessages)
 	if maxErr != nil {
 		writeSQSErrorFromErr(w, maxErr)
@@ -1106,6 +1147,9 @@ func (s *SQSServer) deleteMessage(w http.ResponseWriter, r *http.Request) {
 		writeSQSErrorFromErr(w, err)
 		return
 	}
+	if !s.chargeQueue(w, r, queueName, bucketActionReceive, 1) {
+		return
+	}
 	if err := s.deleteMessageWithRetry(r.Context(), queueName, handle); err != nil {
 		writeSQSErrorFromErr(w, err)
 		return
@@ -1258,6 +1302,9 @@ func (s *SQSServer) changeMessageVisibility(w http.ResponseWriter, r *http.Reque
 		writeSQSErrorFromErr(w, err)
 		return
 	}
+	if !s.chargeQueue(w, r, queueName, bucketActionReceive, 1) {
+		return
+	}
 	if err := s.changeVisibilityWithRetry(r.Context(), queueName, handle, timeout); err != nil {
 		writeSQSErrorFromErr(w, err)
 		return

adapter/sqs_messages.go

@@ -94,6 +94,9 @@ func (s *SQSServer) sendMessageBatch(w http.ResponseWriter, r *http.Request) {
 			"total batch payload exceeds 262144 bytes")
 		return
 	}
+	if !s.chargeQueue(w, r, queueName, bucketActionSend, throttleChargeCount(len(in.Entries))) {
+		return
+	}
 
 	successful, failed, err := s.sendMessageBatchWithRetry(r.Context(), queueName, in.Entries)
 	if err != nil {
@@ -453,6 +456,9 @@ func (s *SQSServer) deleteMessageBatch(w http.ResponseWriter, r *http.Request) {
 		writeSQSErrorFromErr(w, err)
 		return
 	}
+	if !s.chargeQueue(w, r, queueName, bucketActionReceive, throttleChargeCount(len(in.Entries))) {
+		return
+	}
 
 	successful := make([]sqsBatchResultEntry, 0, len(in.Entries))
 	failed := make([]sqsBatchResultErrorEntry, 0)
@@ -519,6 +525,9 @@ func (s *SQSServer) changeMessageVisibilityBatch(w http.ResponseWriter, r *http.
 		writeSQSErrorFromErr(w, err)
 		return
 	}
+	if !s.chargeQueue(w, r, queueName, bucketActionReceive, throttleChargeCount(len(in.Entries))) {
+		return
+	}
 
 	successful := make([]sqsBatchResultEntry, 0, len(in.Entries))
 	failed := make([]sqsBatchResultErrorEntry, 0)