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

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>" }