feat(sqs): finish Milestone 1 - purge, batch, tags, redrive, FIFO, retention reaper (#638)

## Summary

Closes the remaining `NotImplemented` gaps on the SQS adapter so it can
serve real client workloads, not just the catalog and the single-message
hot path.

- **Queue ops**: PurgeQueue (with AWS's 60s cooldown persisted on the
meta record), TagQueue / UntagQueue / ListQueueTags. GetQueueAttributes
now returns ApproximateNumberOfMessages / NotVisible / Delayed,
QueueArn, CreatedTimestamp, LastModifiedTimestamp.
- **Message ops**: SendMessageBatch / DeleteMessageBatch /
ChangeMessageVisibilityBatch with per-entry validation.
MessageAttributes is fully wired: structured input, AWS-canonical MD5
(length-prefixed sorted encoding + type byte), filtered echo on Receive.
- **DLQ redrive**: RedrivePolicy is parsed, persisted, and enforced —
receive transactionally moves a message to the DLQ when it would exceed
maxReceiveCount.
- **FIFO**: per-queue monotonic SequenceNumber, 5-minute dedup window
(`!sqs|msg|dedup|`), and group lock (`!sqs|msg|group|`) that keeps the
head pinned across visibility expiries — released only on Delete / DLQ
redrive / retention.
- **Retention reaper**: new `!sqs|msg|byage|` send-age index plus a
leader-side goroutine that drops expired records (data + vis + byage +
optional group lock) under one OCC dispatch per record. Per-queue budget
prevents starvation; followers short-circuit so only the leader emits
Dispatches.

## Test plan

- [x] `go vet ./...`
- [x] `go test -run 'TestSQS' ./adapter/...`
- [x] 11 new integration tests covering purge cooldown, batch shape
errors, MD5 round-trip, malformed attributes, DLQ redrive, FIFO sequence
/ dedup / content-based dedup, FIFO group lock pinning across visibility
expiry, retention reaper.
- [x] `golangci-lint run --config=.golangci.yaml ./adapter/...` clean.

## Out of scope

- Long-poll cross-node notifier (still polls every 200ms; design §7.3
considers polling acceptable for Milestone 1).
- Operator console UI (`adapter/console.go`).
- Jepsen workload (`jepsen/sqs/`).
- Query-protocol XML compatibility (JSON-1.0 only).

Author: bootjp

adapter/sqs.go

@@ -52,7 +52,6 @@ const (
 // "Common Errors" page of the SQS API reference.
 const (
 	sqsErrInvalidAction      = "InvalidAction"
-	sqsErrNotImplemented     = "NotImplemented"
 	sqsErrInternalFailure    = "InternalFailure"
 	sqsErrServiceUnavailable = "ServiceUnavailable"
 	sqsErrMalformedRequest   = "MalformedQueryString"
@@ -69,6 +68,14 @@ type SQSServer struct {
 	leaderSQS      map[string]string
 	region         string
 	staticCreds    map[string]string
+	// reaperCtx / reaperCancel drive the retention sweeper goroutine.
+	// Both are initialized in NewSQSServer (never reassigned) so a
+	// concurrent Stop() that lands before Run() completes still reads
+	// a stable cancel func — unlike a Run-time assignment, which the
+	// race detector flagged because Run and Stop run on different
+	// goroutines without ordering between them.
+	reaperCtx    context.Context
+	reaperCancel context.CancelFunc
 }
 
 // WithSQSLeaderMap configures the Raft-address-to-SQS-address mapping used to
@@ -84,10 +91,13 @@ func WithSQSLeaderMap(m map[string]string) SQSServerOption {
 }
 
 func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordinator, opts ...SQSServerOption) *SQSServer {
+	reaperCtx, reaperCancel := context.WithCancel(context.Background())
 	s := &SQSServer{
-		listen:      listen,
-		store:       st,
-		coordinator: coordinate,
+		listen:       listen,
+		store:        st,
+		coordinator:  coordinate,
+		reaperCtx:    reaperCtx,
+		reaperCancel: reaperCancel,
 	}
 	s.targetHandlers = map[string]func(http.ResponseWriter, *http.Request){
 		sqsCreateQueueTarget:               s.createQueue,
@@ -96,17 +106,17 @@ func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordin
 		sqsGetQueueUrlTarget:               s.getQueueUrl,
 		sqsGetQueueAttributesTarget:        s.getQueueAttributes,
 		sqsSetQueueAttributesTarget:        s.setQueueAttributes,
-		sqsPurgeQueueTarget:                s.notImplemented("PurgeQueue"),
+		sqsPurgeQueueTarget:                s.purgeQueue,
 		sqsSendMessageTarget:               s.sendMessage,
-		sqsSendMessageBatchTarget:          s.notImplemented("SendMessageBatch"),
+		sqsSendMessageBatchTarget:          s.sendMessageBatch,
 		sqsReceiveMessageTarget:            s.receiveMessage,
 		sqsDeleteMessageTarget:             s.deleteMessage,
-		sqsDeleteMessageBatchTarget:        s.notImplemented("DeleteMessageBatch"),
+		sqsDeleteMessageBatchTarget:        s.deleteMessageBatch,
 		sqsChangeMessageVisibilityTarget:   s.changeMessageVisibility,
-		sqsChangeMessageVisibilityBatchTgt: s.notImplemented("ChangeMessageVisibilityBatch"),
-		sqsTagQueueTarget:                  s.notImplemented("TagQueue"),
-		sqsUntagQueueTarget:                s.notImplemented("UntagQueue"),
-		sqsListQueueTagsTarget:             s.notImplemented("ListQueueTags"),
+		sqsChangeMessageVisibilityBatchTgt: s.changeMessageVisibilityBatch,
+		sqsTagQueueTarget:                  s.tagQueue,
+		sqsUntagQueueTarget:                s.untagQueue,
+		sqsListQueueTagsTarget:             s.listQueueTags,
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", s.handle)
@@ -120,13 +130,17 @@ func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordin
 }
 
 func (s *SQSServer) Run() error {
+	s.startReaper(s.reaperCtx)
 	if err := s.httpServer.Serve(s.listen); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		return errors.WithStack(err)
 	}
 	return nil
 }
 
 func (s *SQSServer) Stop() {
+	if s.reaperCancel != nil {
+		s.reaperCancel()
+	}
 	if s.httpServer != nil {
 		_ = s.httpServer.Shutdown(context.Background())
 	}
@@ -237,15 +251,6 @@ func sqsLeaderProxyErrorWriter(w http.ResponseWriter, status int, message string
 	writeSQSError(w, status, sqsErrServiceUnavailable, message)
 }
 
-// notImplemented returns a handler that responds with a JSON-protocol
-// NotImplemented error so clients get a clean signal while the real handlers
-// are still being built out.
-func (s *SQSServer) notImplemented(op string) func(http.ResponseWriter, *http.Request) {
-	return func(w http.ResponseWriter, _ *http.Request) {
-		writeSQSError(w, http.StatusNotImplemented, sqsErrNotImplemented, op+" is not implemented yet")
-	}
-}
-
 // writeSQSError emits an SQS JSON-protocol error envelope. AWS returns:
 //
 //	{ "__type": "<code>", "message": "<text>" }