From 70595c0db181d69d1e5177bf5e646e2618c0e57e Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Sat, 18 Jul 2026 08:10:38 +0900
Subject: [PATCH 1/6] feat(sqs): add dead-letter redrive ops for full 23/23
 coverage

Implement the four remaining SQS operations, all in the dead-letter
redrive family:

- ListDeadLetterSourceQueues
- StartMessageMoveTask
- ListMessageMoveTasks
- CancelMessageMoveTask

The store gains a move-task registry and a source-queue lookup that
reuse the existing RedrivePolicy fields (MaxReceiveCount /
DeadLetterTargetArn). Redrive runs synchronously in the in-memory store
and drains one queue lock at a time to stay deadlock-free with the
source->DLQ path in ReceiveMessage.

Wire field casing matches the botocore SQS JSON model (queueUrls is
lowercase for ListDeadLetterSourceQueues; the rest PascalCase).

Adds store unit tests and boto3 compat tests, and corrects the stale
SQS docs (batch/DLQ/FIFO/message-attributes were wrongly listed as
unsupported).
---
 docs/services-matrix.md             |   2 +-
 docs/services/sqs.md                |  29 +++--
 internal/services/sqs/provider.go   |  89 +++++++++++++++
 internal/services/sqs/store.go      | 169 +++++++++++++++++++++++++++-
 internal/services/sqs/store_test.go |  79 +++++++++++++
 tests/compatibility/test_sqs.py     |  46 ++++++++
 6 files changed, 397 insertions(+), 17 deletions(-)

diff --git a/docs/services-matrix.md b/docs/services-matrix.md
index e59a5959..870d7e3e 100644
--- a/docs/services-matrix.md
+++ b/docs/services-matrix.md
@@ -12,7 +12,7 @@ DevCloud is a Go-based local cloud environment with AWS API compatibility. This
 
 | Tier | Services | Ops | Description |
 |------|----------|-----|-------------|
-| Tier 1 (Big 6) | S3, SQS, DynamoDB, Lambda, IAM, STS | 124+ | Core services |
+| Tier 1 (Big 6) | S3, SQS, DynamoDB, Lambda, IAM, STS | 128+ | Core services (SQS at full 23/23 op coverage) |
 | Tier 2 (Integration) | EventBridge, SNS, CW Logs, CloudWatch, KMS, Secrets Manager, SSM, ECR | 157+ | Integration services |
 | Tier 3 (Extended) | EFS, EBS, EC2, Route53, ACM, ECS, Bedrock, Account, Pipes, CloudControl, RGTAPI, AppAutoScaling, Firehose, S3Tables, MWAA, Scheduler, Support, IdentityStore, MediaConvert, Textract, ServerlessRepo, DDB Streams, SFN, Kinesis, CloudFormation | 900+ | Extended platform services, networking, and services requiring custom integration logic |
 | Category Expansion | 40+ additional services | 2,000+ | All remaining AWS services scaffolded from Smithy models |
diff --git a/docs/services/sqs.md b/docs/services/sqs.md
index 6c2c12a2..942745ed 100644
--- a/docs/services/sqs.md
+++ b/docs/services/sqs.md
@@ -8,15 +8,24 @@ SQS supports dual protocol: legacy Query protocol (form-encoded with XML respons
 
 ## Supported APIs
 
+All 23 SQS operations are handled over the JSON 1.0 protocol.
+
 | Operation | Description |
 |-----------|-------------|
-| CreateQueue | Create a new queue (returns existing if name matches) |
+| CreateQueue | Create a new queue (returns existing if name matches); supports FIFO + attributes |
 | ListQueues | List queues with optional name prefix filter |
 | GetQueueUrl | Resolve queue name to its URL |
-| SendMessage | Send a message to a queue (computes MD5 hash) |
-| ReceiveMessage | Receive messages with visibility timeout and MaxNumberOfMessages |
-| DeleteMessage | Delete a message by receipt handle |
 | DeleteQueue | Delete a queue and all its messages |
+| PurgeQueue | Remove all messages from a queue |
+| SendMessage / SendMessageBatch | Send one or many messages (computes MD5 hash) |
+| ReceiveMessage | Receive messages with visibility timeout and MaxNumberOfMessages |
+| DeleteMessage / DeleteMessageBatch | Delete one or many messages by receipt handle |
+| ChangeMessageVisibility / ChangeMessageVisibilityBatch | Adjust message visibility timeout |
+| GetQueueAttributes / SetQueueAttributes | Read/write queue attributes (incl. RedrivePolicy) |
+| TagQueue / UntagQueue / ListQueueTags | Manage queue tags |
+| ListDeadLetterSourceQueues | List queues that use a queue as their dead-letter target |
+| StartMessageMoveTask / ListMessageMoveTasks / CancelMessageMoveTask | Redrive messages out of a dead-letter queue |
+| AddPermission / RemovePermission | Accepted (no-op — permissions are not enforced) |
 
 ## boto3 Examples
 
@@ -74,10 +83,8 @@ aws --endpoint-url http://localhost:4747 sqs receive-message \
 ## Known Limitations
 
 - In-memory only — no persistence across restarts
-- No queue attributes (DelaySeconds, VisibilityTimeout at queue level, etc.)
-- No batch operations (SendMessageBatch, DeleteMessageBatch)
-- No dead-letter queues
-- No FIFO queues
-- No message retention/expiry
-- No message attributes
-- No long polling (WaitTimeSeconds)
+- Dead-letter redrive (message move tasks) runs synchronously: tasks complete immediately,
+  so there is no `RUNNING` status and `MaxNumberOfMessagesPerSecond` is not throttled
+- `AddPermission`/`RemovePermission` are accepted but not enforced (no access-policy checks)
+- No message retention/expiry enforcement
+- No long polling (`WaitTimeSeconds` is not honored — receives return immediately)
diff --git a/internal/services/sqs/provider.go b/internal/services/sqs/provider.go
index 504de79c..09873e1e 100644
--- a/internal/services/sqs/provider.go
+++ b/internal/services/sqs/provider.go
@@ -166,6 +166,15 @@ func (p *SQSProvider) handleJSONRequest(op string, req *http.Request) (*plugin.R
 		return p.untagQueueJSON(params)
 	case "ListQueueTags":
 		return p.listQueueTagsJSON(params)
+	// Dead-letter redrive
+	case "ListDeadLetterSourceQueues":
+		return p.listDeadLetterSourceQueuesJSON(params)
+	case "StartMessageMoveTask":
+		return p.startMessageMoveTaskJSON(params)
+	case "ListMessageMoveTasks":
+		return p.listMessageMoveTasksJSON(params)
+	case "CancelMessageMoveTask":
+		return p.cancelMessageMoveTaskJSON(params)
 	// Permissions (stubs)
 	case "AddPermission":
 		return jsonResp(http.StatusOK, map[string]any{})
@@ -602,6 +611,86 @@ func (p *SQSProvider) purgeQueueJSON(params map[string]any) (*plugin.Response, e
 	return jsonResp(http.StatusOK, map[string]any{})
 }
 
+func (p *SQSProvider) listDeadLetterSourceQueuesJSON(params map[string]any) (*plugin.Response, error) {
+	queueURL := strParam(params, "QueueUrl")
+	if queueURL == "" {
+		return jsonError("MissingParameter", "QueueUrl is required", http.StatusBadRequest), nil
+	}
+
+	name := queueNameFromURL(queueURL)
+	urls, err := p.store.ListDeadLetterSourceQueues(name, defaultAccountID)
+	if err != nil {
+		return jsonError("AWS.SimpleQueueService.NonExistentQueue", "queue not found", http.StatusBadRequest), nil
+	}
+
+	// Wire member name is lowercase "queueUrls" for this op (per the SQS JSON model),
+	// unlike ListQueues' "QueueUrls".
+	return jsonResp(http.StatusOK, map[string]any{"queueUrls": urls})
+}
+
+func (p *SQSProvider) startMessageMoveTaskJSON(params map[string]any) (*plugin.Response, error) {
+	sourceArn := strParam(params, "SourceArn")
+	if sourceArn == "" {
+		return jsonError("MissingParameter", "SourceArn is required", http.StatusBadRequest), nil
+	}
+	destArn := strParam(params, "DestinationArn")
+	maxRate := intParam(params, "MaxNumberOfMessagesPerSecond", 0)
+
+	handle, err := p.store.StartMessageMoveTask(sourceArn, destArn, maxRate, defaultAccountID)
+	switch {
+	case err == ErrQueueNotFound:
+		return jsonError("ResourceNotFoundException", "queue not found", http.StatusBadRequest), nil
+	case err == ErrSourceNotDLQ:
+		return jsonError("AWS.SimpleQueueService.UnsupportedOperation",
+			"source queue must be configured as a dead-letter queue", http.StatusBadRequest), nil
+	case err != nil:
+		return nil, err
+	}
+
+	return jsonResp(http.StatusOK, map[string]any{"TaskHandle": handle})
+}
+
+func (p *SQSProvider) listMessageMoveTasksJSON(params map[string]any) (*plugin.Response, error) {
+	sourceArn := strParam(params, "SourceArn")
+	if sourceArn == "" {
+		return jsonError("MissingParameter", "SourceArn is required", http.StatusBadRequest), nil
+	}
+	maxResults := intParam(params, "MaxResults", 0)
+
+	tasks := p.store.ListMessageMoveTasks(sourceArn, maxResults)
+	results := make([]map[string]any, 0, len(tasks))
+	for _, t := range tasks {
+		entry := map[string]any{
+			"TaskHandle":                       t.Handle,
+			"Status":                           t.Status,
+			"SourceArn":                        t.SourceArn,
+			"MaxNumberOfMessagesPerSecond":     t.MaxRate,
+			"ApproximateNumberOfMessagesMoved": t.Moved,
+			"StartedTimestamp":                 t.StartedUnix,
+		}
+		if t.DestinationArn != "" {
+			entry["DestinationArn"] = t.DestinationArn
+		}
+		results = append(results, entry)
+	}
+
+	return jsonResp(http.StatusOK, map[string]any{"Results": results})
+}
+
+func (p *SQSProvider) cancelMessageMoveTaskJSON(params map[string]any) (*plugin.Response, error) {
+	handle := strParam(params, "TaskHandle")
+	if handle == "" {
+		return jsonError("MissingParameter", "TaskHandle is required", http.StatusBadRequest), nil
+	}
+
+	moved, err := p.store.CancelMessageMoveTask(handle)
+	if err != nil {
+		return jsonError("ResourceNotFoundException", "task not found", http.StatusBadRequest), nil
+	}
+
+	return jsonResp(http.StatusOK, map[string]any{"ApproximateNumberOfMessagesMoved": moved})
+}
+
 // parseJSONMessageAttributes extracts MessageAttributes from JSON params.
 func parseJSONMessageAttributes(params map[string]any) map[string]MessageAttribute {
 	raw, ok := params["MessageAttributes"]
diff --git a/internal/services/sqs/store.go b/internal/services/sqs/store.go
index bbdfe094..6dfa8070 100644
--- a/internal/services/sqs/store.go
+++ b/internal/services/sqs/store.go
@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -20,6 +21,8 @@ var (
 	ErrQueueAlreadyExists = errors.New("queue already exists")
 	ErrQueueNotFound      = errors.New("queue not found")
 	ErrReceiptInvalid     = errors.New("receipt handle is invalid")
+	ErrSourceNotDLQ       = errors.New("source queue is not configured as a dead-letter queue")
+	ErrTaskNotFound       = errors.New("message move task not found")
 )
 
 // QueueInfo holds metadata about an SQS queue.
@@ -82,9 +85,10 @@ type queue struct {
 
 // QueueStore is a thread-safe in-memory store for SQS queues.
 type QueueStore struct {
-	mu     sync.RWMutex
-	queues map[string]*queue // key: "accountID/queueName"
-	port   int               // HTTP port for queue-URL construction
+	mu        sync.RWMutex
+	queues    map[string]*queue    // key: "accountID/queueName"
+	moveTasks map[string]*moveTask // key: task handle
+	port      int                  // HTTP port for queue-URL construction
 }
 
 // NewQueueStore creates and returns an empty QueueStore bound to the given
@@ -95,8 +99,9 @@ func NewQueueStore(port int) *QueueStore {
 		port = 4747
 	}
 	return &QueueStore{
-		queues: make(map[string]*queue),
-		port:   port,
+		queues:    make(map[string]*queue),
+		moveTasks: make(map[string]*moveTask),
+		port:      port,
 	}
 }
 
@@ -656,3 +661,157 @@ func (s *QueueStore) ListQueueTags(queueName, accountID string) (map[string]stri
 	}
 	return result, nil
 }
+
+// --- Dead-letter redrive (message move tasks) ---
+
+// moveTask records a dead-letter redrive task. Redrive runs synchronously (in-memory),
+// so a task is created already in a terminal state.
+type moveTask struct {
+	Handle         string
+	SourceArn      string
+	DestinationArn string
+	MaxRate        int    // MaxNumberOfMessagesPerSecond; 0 = unlimited
+	Status         string // "COMPLETED" | "CANCELLED"
+	Moved          int
+	StartedUnix    int64
+}
+
+// sourceQueuesForDLQ returns the queues under accountID that name dlqName as their
+// dead-letter target. The caller must hold s.mu.
+func (s *QueueStore) sourceQueuesForDLQ(dlqName, accountID string) []*queue {
+	var out []*queue
+	for _, q := range s.queues {
+		if q.info.AccountID != accountID {
+			continue
+		}
+		q.mu.Lock()
+		target := arnToQueueName(q.DeadLetterTargetArn)
+		q.mu.Unlock()
+		if target == dlqName {
+			out = append(out, q)
+		}
+	}
+	return out
+}
+
+// ListDeadLetterSourceQueues returns the URLs of queues that use the named queue as
+// their dead-letter target. Returns ErrQueueNotFound if the DLQ does not exist.
+func (s *QueueStore) ListDeadLetterSourceQueues(dlqName, accountID string) ([]string, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	if _, ok := s.queues[queueKey(accountID, dlqName)]; !ok {
+		return nil, ErrQueueNotFound
+	}
+
+	sources := s.sourceQueuesForDLQ(dlqName, accountID)
+	urls := make([]string, 0, len(sources))
+	for _, q := range sources {
+		urls = append(urls, q.info.URL)
+	}
+	return urls, nil
+}
+
+// StartMessageMoveTask redrives messages out of a dead-letter queue (identified by
+// sourceArn) into a destination queue. When destArn is empty, messages go to the first
+// queue that references the DLQ as its dead-letter target. Returns the task handle.
+func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int, accountID string) (string, error) {
+	srcName := arnToQueueName(sourceArn)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	src, ok := s.queues[queueKey(accountID, srcName)]
+	if !ok {
+		return "", ErrQueueNotFound
+	}
+
+	sources := s.sourceQueuesForDLQ(srcName, accountID)
+	if len(sources) == 0 {
+		return "", ErrSourceNotDLQ
+	}
+
+	dst := sources[0]
+	// ponytail: per-message origin isn't tracked, so multi-source redrive without an
+	// explicit DestinationArn goes to the first referencing source; pass DestinationArn
+	// for exact routing.
+	if destArn != "" {
+		dst, ok = s.queues[queueKey(accountID, arnToQueueName(destArn))]
+		if !ok {
+			return "", ErrQueueNotFound
+		}
+	}
+
+	// ponytail: redrive runs synchronously with no RUNNING state or rate limiting;
+	// upgrade to a goroutine + MaxRate throttle if timing fidelity is ever needed.
+	// Lock only one queue at a time (drain the source, then fill the destination) to stay
+	// deadlock-free with the source→DLQ path in ReceiveMessage.
+	src.mu.Lock()
+	moved := make([]*Message, 0, len(src.messages))
+	for _, m := range src.messages {
+		if !m.deleted {
+			moved = append(moved, m)
+		}
+	}
+	src.messages = make([]*Message, 0)
+	src.mu.Unlock()
+
+	dst.mu.Lock()
+	for _, m := range moved {
+		m.ReceiptHandle = randomID(32)
+		m.invisibleUntil = time.Time{}
+		m.receiveCount = 0
+		m.deleted = false
+		dst.messages = append(dst.messages, m)
+	}
+	dst.mu.Unlock()
+
+	handle := randomID(16)
+	s.moveTasks[handle] = &moveTask{
+		Handle:         handle,
+		SourceArn:      sourceArn,
+		DestinationArn: destArn,
+		MaxRate:        maxRate,
+		Status:         "COMPLETED",
+		Moved:          len(moved),
+		StartedUnix:    time.Now().Unix(),
+	}
+	return handle, nil
+}
+
+// ListMessageMoveTasks returns move tasks for the given source ARN, newest first,
+// capped at maxResults (default 10 when maxResults <= 0).
+func (s *QueueStore) ListMessageMoveTasks(sourceArn string, maxResults int) []*moveTask {
+	if maxResults <= 0 {
+		maxResults = 10
+	}
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	var out []*moveTask
+	for _, t := range s.moveTasks {
+		if t.SourceArn == sourceArn {
+			out = append(out, t)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].StartedUnix > out[j].StartedUnix })
+	if len(out) > maxResults {
+		out = out[:maxResults]
+	}
+	return out
+}
+
+// CancelMessageMoveTask cancels the task with the given handle and returns the number of
+// messages already moved. Returns ErrTaskNotFound if the handle is unknown.
+func (s *QueueStore) CancelMessageMoveTask(handle string) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	t, ok := s.moveTasks[handle]
+	if !ok {
+		return 0, ErrTaskNotFound
+	}
+	t.Status = "CANCELLED"
+	return t.Moved, nil
+}
diff --git a/internal/services/sqs/store_test.go b/internal/services/sqs/store_test.go
index ab7a6f2d..4150f6d7 100644
--- a/internal/services/sqs/store_test.go
+++ b/internal/services/sqs/store_test.go
@@ -168,6 +168,85 @@ func TestQueueStore_DLQ(t *testing.T) {
 	assert.Equal(t, "dlq-test", dlqMsgs[0].Body)
 }
 
+func TestQueueStore_ListDeadLetterSourceQueues(t *testing.T) {
+	store := NewQueueStore(0)
+	require.NoError(t, store.CreateQueue("mmt-source", testAccount))
+	require.NoError(t, store.CreateQueue("mmt-dlq", testAccount))
+
+	dlqArn := fmt.Sprintf("arn:aws:sqs:us-east-1:%s:mmt-dlq", testAccount)
+	require.NoError(t, store.SetQueueAttributes("mmt-source", testAccount, map[string]string{
+		"RedrivePolicy": `{"maxReceiveCount":1,"deadLetterTargetArn":"` + dlqArn + `"}`,
+	}))
+
+	// The DLQ lists its source queue.
+	sources, err := store.ListDeadLetterSourceQueues("mmt-dlq", testAccount)
+	require.NoError(t, err)
+	require.Len(t, sources, 1)
+	assert.Contains(t, sources[0], "mmt-source")
+
+	// A queue that is not used as a DLQ has no source queues.
+	empty, err := store.ListDeadLetterSourceQueues("mmt-source", testAccount)
+	require.NoError(t, err)
+	assert.Len(t, empty, 0)
+
+	// An unknown queue errors.
+	_, err = store.ListDeadLetterSourceQueues("no-such-queue", testAccount)
+	assert.ErrorIs(t, err, ErrQueueNotFound)
+}
+
+func TestQueueStore_MessageMoveTaskLifecycle(t *testing.T) {
+	store := NewQueueStore(0)
+	require.NoError(t, store.CreateQueue("redrive-source", testAccount))
+	require.NoError(t, store.CreateQueue("redrive-dlq", testAccount))
+
+	dlqArn := fmt.Sprintf("arn:aws:sqs:us-east-1:%s:redrive-dlq", testAccount)
+	require.NoError(t, store.SetQueueAttributes("redrive-source", testAccount, map[string]string{
+		"RedrivePolicy": `{"maxReceiveCount":1,"deadLetterTargetArn":"` + dlqArn + `"}`,
+	}))
+
+	_, err := store.SendMessage("redrive-source", testAccount, "redrive-me")
+	require.NoError(t, err)
+
+	// Exceed maxReceiveCount so the message lands in the DLQ.
+	for i := 0; i < 2; i++ {
+		_, rerr := store.ReceiveMessage("redrive-source", testAccount, 1, 0)
+		require.NoError(t, rerr)
+	}
+	dlqMsgs, err := store.ReceiveMessage("redrive-dlq", testAccount, 1, 0)
+	require.NoError(t, err)
+	require.Len(t, dlqMsgs, 1)
+
+	// Redrive back to the source (no explicit destination).
+	handle, err := store.StartMessageMoveTask(dlqArn, "", 0, testAccount)
+	require.NoError(t, err)
+	require.NotEmpty(t, handle)
+
+	tasks := store.ListMessageMoveTasks(dlqArn, 0)
+	require.Len(t, tasks, 1)
+	assert.Equal(t, "COMPLETED", tasks[0].Status)
+	assert.Equal(t, 1, tasks[0].Moved)
+
+	// The message is back on the source queue.
+	back, err := store.ReceiveMessage("redrive-source", testAccount, 1, 0)
+	require.NoError(t, err)
+	require.Len(t, back, 1)
+	assert.Equal(t, "redrive-me", back[0].Body)
+
+	// Cancel reports the moved count.
+	moved, err := store.CancelMessageMoveTask(handle)
+	require.NoError(t, err)
+	assert.Equal(t, 1, moved)
+
+	// An unknown handle errors.
+	_, err = store.CancelMessageMoveTask("bogus-handle")
+	assert.ErrorIs(t, err, ErrTaskNotFound)
+
+	// A queue that is not a DLQ cannot start a move task.
+	srcArn := fmt.Sprintf("arn:aws:sqs:us-east-1:%s:redrive-source", testAccount)
+	_, err = store.StartMessageMoveTask(srcArn, "", 0, testAccount)
+	assert.ErrorIs(t, err, ErrSourceNotDLQ)
+}
+
 func TestQueueStore_Tags(t *testing.T) {
 	store := NewQueueStore(0)
 	require.NoError(t, store.CreateQueue("tag-queue", testAccount))
diff --git a/tests/compatibility/test_sqs.py b/tests/compatibility/test_sqs.py
index fdbbe4f2..67b421f0 100644
--- a/tests/compatibility/test_sqs.py
+++ b/tests/compatibility/test_sqs.py
@@ -1,3 +1,5 @@
+import json
+
 import pytest
 from botocore.exceptions import ClientError
 
@@ -193,3 +195,47 @@ def test_fifo_dedup(sqs_client):
     )
     msgs = sqs_client.receive_message(QueueUrl=url, MaxNumberOfMessages=10)
     assert len(msgs["Messages"]) == 1
+
+
+def _redrive_pair(sqs_client, src_name, dlq_name):
+    """Create a source queue wired to a DLQ (maxReceiveCount=1); return (src_url, dlq_url, dlq_arn)."""
+    src = sqs_client.create_queue(QueueName=src_name)["QueueUrl"]
+    dlq = sqs_client.create_queue(QueueName=dlq_name)["QueueUrl"]
+    dlq_arn = sqs_client.get_queue_attributes(
+        QueueUrl=dlq, AttributeNames=["QueueArn"]
+    )["Attributes"]["QueueArn"]
+    sqs_client.set_queue_attributes(
+        QueueUrl=src,
+        Attributes={
+            "RedrivePolicy": json.dumps(
+                {"deadLetterTargetArn": dlq_arn, "maxReceiveCount": 1}
+            )
+        },
+    )
+    return src, dlq, dlq_arn
+
+
+def test_list_dead_letter_source_queues(sqs_client):
+    src, dlq, _ = _redrive_pair(sqs_client, "dlq-src", "dlq-target")
+    resp = sqs_client.list_dead_letter_source_queues(QueueUrl=dlq)
+    urls = resp["queueUrls"]
+    assert any("dlq-src" in u for u in urls)
+
+
+def test_message_move_task_flow(sqs_client):
+    src, dlq, dlq_arn = _redrive_pair(sqs_client, "mmt-src", "mmt-dlq")
+
+    sqs_client.send_message(QueueUrl=src, MessageBody="redrive-me")
+    # Exceed maxReceiveCount so the message is moved to the DLQ.
+    sqs_client.receive_message(QueueUrl=src, VisibilityTimeout=0)
+    sqs_client.receive_message(QueueUrl=src, VisibilityTimeout=0)
+
+    handle = sqs_client.start_message_move_task(SourceArn=dlq_arn)["TaskHandle"]
+    assert handle
+
+    tasks = sqs_client.list_message_move_tasks(SourceArn=dlq_arn)["Results"]
+    assert len(tasks) >= 1
+    assert tasks[0]["SourceArn"] == dlq_arn
+
+    cancel = sqs_client.cancel_message_move_task(TaskHandle=handle)
+    assert "ApproximateNumberOfMessagesMoved" in cancel

From dd21f4fb262b227f640151535a7e69503e1dce44 Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Sat, 18 Jul 2026 08:11:18 +0900
Subject: [PATCH 2/6] docs(changelog): add changie entry for SQS redrive ops
 (#94)

---
 changes/unreleased/Added-20260718-081112.yaml | 5 +++++
 1 file changed, 5 insertions(+)
 create mode 100644 changes/unreleased/Added-20260718-081112.yaml

diff --git a/changes/unreleased/Added-20260718-081112.yaml b/changes/unreleased/Added-20260718-081112.yaml
new file mode 100644
index 00000000..a852ff8e
--- /dev/null
+++ b/changes/unreleased/Added-20260718-081112.yaml
@@ -0,0 +1,5 @@
+kind: Added
+body: Add SQS dead-letter redrive operations (ListDeadLetterSourceQueues, StartMessageMoveTask, ListMessageMoveTasks, CancelMessageMoveTask) — SQS now covers all 23 operations
+time: 2026-07-18T08:11:12.483907+09:00
+custom:
+    Issue: "94"

From 16390653d3e6a53a56117694922a131742577cad Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Sat, 18 Jul 2026 08:19:41 +0900
Subject: [PATCH 3/6] fix(sqs): bound move-task registry and make multi-source
 redrive explicit

Address code review on the redrive ops:

- StartMessageMoveTask now keeps only the latest task per source, so the
  moveTasks map can't grow without bound.
- When a DLQ has multiple source queues and no DestinationArn is given,
  redrive is rejected (ErrAmbiguousDest -> InvalidParameterValue) instead
  of silently routing to the first source; single-source still defaults.

Expand tests: store tests for the multi-source ambiguity, explicit-dest
routing, and the bounded registry; compat tests for empty/non-existent
list_dead_letter_source_queues, destination routing, ambiguous dest,
unsupported (non-DLQ) source, and cancel missing/unknown handle. Note
that omitting a required member (SourceArn) raises botocore's
ParamValidationError client-side, so that path is asserted as such.
---
 internal/services/sqs/provider.go   |   4 ++
 internal/services/sqs/store.go      |  22 ++++--
 internal/services/sqs/store_test.go |  49 +++++++++++++
 tests/compatibility/test_sqs.py     | 107 +++++++++++++++++++++++++---
 4 files changed, 169 insertions(+), 13 deletions(-)

diff --git a/internal/services/sqs/provider.go b/internal/services/sqs/provider.go
index 09873e1e..d61e2be8 100644
--- a/internal/services/sqs/provider.go
+++ b/internal/services/sqs/provider.go
@@ -643,6 +643,10 @@ func (p *SQSProvider) startMessageMoveTaskJSON(params map[string]any) (*plugin.R
 	case err == ErrSourceNotDLQ:
 		return jsonError("AWS.SimpleQueueService.UnsupportedOperation",
 			"source queue must be configured as a dead-letter queue", http.StatusBadRequest), nil
+	case err == ErrAmbiguousDest:
+		return jsonError("InvalidParameterValue",
+			"DestinationArn is required when the source queue has multiple dead-letter source queues",
+			http.StatusBadRequest), nil
 	case err != nil:
 		return nil, err
 	}
diff --git a/internal/services/sqs/store.go b/internal/services/sqs/store.go
index 6dfa8070..3d1f080a 100644
--- a/internal/services/sqs/store.go
+++ b/internal/services/sqs/store.go
@@ -23,6 +23,7 @@ var (
 	ErrReceiptInvalid     = errors.New("receipt handle is invalid")
 	ErrSourceNotDLQ       = errors.New("source queue is not configured as a dead-letter queue")
 	ErrTaskNotFound       = errors.New("message move task not found")
+	ErrAmbiguousDest      = errors.New("destinationArn is required when the dead-letter queue has multiple source queues")
 )
 
 // QueueInfo holds metadata about an SQS queue.
@@ -731,15 +732,20 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 		return "", ErrSourceNotDLQ
 	}
 
-	dst := sources[0]
-	// ponytail: per-message origin isn't tracked, so multi-source redrive without an
-	// explicit DestinationArn goes to the first referencing source; pass DestinationArn
-	// for exact routing.
+	var dst *queue
 	if destArn != "" {
 		dst, ok = s.queues[queueKey(accountID, arnToQueueName(destArn))]
 		if !ok {
 			return "", ErrQueueNotFound
 		}
+	} else {
+		// Without an explicit destination, messages go back to the source queue. Per-message
+		// origin isn't tracked, so this is only unambiguous when the DLQ has a single source;
+		// with multiple sources the caller must pass DestinationArn.
+		if len(sources) > 1 {
+			return "", ErrAmbiguousDest
+		}
+		dst = sources[0]
 	}
 
 	// ponytail: redrive runs synchronously with no RUNNING state or rate limiting;
@@ -766,6 +772,14 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 	}
 	dst.mu.Unlock()
 
+	// Keep only the latest task per source so the registry stays bounded (a source can have
+	// at most one task record); a new task supersedes any prior one for the same source.
+	for h, t := range s.moveTasks {
+		if t.SourceArn == sourceArn {
+			delete(s.moveTasks, h)
+		}
+	}
+
 	handle := randomID(16)
 	s.moveTasks[handle] = &moveTask{
 		Handle:         handle,
diff --git a/internal/services/sqs/store_test.go b/internal/services/sqs/store_test.go
index 4150f6d7..21a33b7a 100644
--- a/internal/services/sqs/store_test.go
+++ b/internal/services/sqs/store_test.go
@@ -247,6 +247,55 @@ func TestQueueStore_MessageMoveTaskLifecycle(t *testing.T) {
 	assert.ErrorIs(t, err, ErrSourceNotDLQ)
 }
 
+func TestQueueStore_MessageMoveTaskMultiSource(t *testing.T) {
+	store := NewQueueStore(0)
+	require.NoError(t, store.CreateQueue("multi-src1", testAccount))
+	require.NoError(t, store.CreateQueue("multi-src2", testAccount))
+	require.NoError(t, store.CreateQueue("multi-dlq", testAccount))
+
+	dlqArn := fmt.Sprintf("arn:aws:sqs:us-east-1:%s:multi-dlq", testAccount)
+	redrive := `{"maxReceiveCount":1,"deadLetterTargetArn":"` + dlqArn + `"}`
+	require.NoError(t, store.SetQueueAttributes("multi-src1", testAccount, map[string]string{"RedrivePolicy": redrive}))
+	require.NoError(t, store.SetQueueAttributes("multi-src2", testAccount, map[string]string{"RedrivePolicy": redrive}))
+
+	_, err := store.SendMessage("multi-dlq", testAccount, "orphan")
+	require.NoError(t, err)
+
+	// Two sources share the DLQ, so redrive without a destination is ambiguous.
+	_, err = store.StartMessageMoveTask(dlqArn, "", 0, testAccount)
+	assert.ErrorIs(t, err, ErrAmbiguousDest)
+
+	// An explicit destination routes deterministically.
+	dstArn := fmt.Sprintf("arn:aws:sqs:us-east-1:%s:multi-src2", testAccount)
+	handle, err := store.StartMessageMoveTask(dlqArn, dstArn, 0, testAccount)
+	require.NoError(t, err)
+	require.NotEmpty(t, handle)
+
+	msgs, err := store.ReceiveMessage("multi-src2", testAccount, 1, 0)
+	require.NoError(t, err)
+	require.Len(t, msgs, 1)
+	assert.Equal(t, "orphan", msgs[0].Body)
+}
+
+func TestQueueStore_MessageMoveTaskRegistryBounded(t *testing.T) {
+	store := NewQueueStore(0)
+	require.NoError(t, store.CreateQueue("bound-src", testAccount))
+	require.NoError(t, store.CreateQueue("bound-dlq", testAccount))
+
+	dlqArn := fmt.Sprintf("arn:aws:sqs:us-east-1:%s:bound-dlq", testAccount)
+	require.NoError(t, store.SetQueueAttributes("bound-src", testAccount, map[string]string{
+		"RedrivePolicy": `{"maxReceiveCount":1,"deadLetterTargetArn":"` + dlqArn + `"}`,
+	}))
+
+	// Starting many tasks for the same source keeps only the latest record.
+	for i := 0; i < 5; i++ {
+		_, err := store.StartMessageMoveTask(dlqArn, "", 0, testAccount)
+		require.NoError(t, err)
+	}
+	assert.Len(t, store.moveTasks, 1)
+	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 0), 1)
+}
+
 func TestQueueStore_Tags(t *testing.T) {
 	store := NewQueueStore(0)
 	require.NoError(t, store.CreateQueue("tag-queue", testAccount))
diff --git a/tests/compatibility/test_sqs.py b/tests/compatibility/test_sqs.py
index 67b421f0..3a29d8d1 100644
--- a/tests/compatibility/test_sqs.py
+++ b/tests/compatibility/test_sqs.py
@@ -1,7 +1,7 @@
 import json
 
 import pytest
-from botocore.exceptions import ClientError
+from botocore.exceptions import ClientError, ParamValidationError
 
 
 def test_create_queue(sqs_client):
@@ -197,21 +197,30 @@ def test_fifo_dedup(sqs_client):
     assert len(msgs["Messages"]) == 1
 
 
-def _redrive_pair(sqs_client, src_name, dlq_name):
-    """Create a source queue wired to a DLQ (maxReceiveCount=1); return (src_url, dlq_url, dlq_arn)."""
-    src = sqs_client.create_queue(QueueName=src_name)["QueueUrl"]
-    dlq = sqs_client.create_queue(QueueName=dlq_name)["QueueUrl"]
-    dlq_arn = sqs_client.get_queue_attributes(
-        QueueUrl=dlq, AttributeNames=["QueueArn"]
-    )["Attributes"]["QueueArn"]
+def _queue_arn(sqs_client, url):
+    return sqs_client.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])[
+        "Attributes"
+    ]["QueueArn"]
+
+
+def _wire_to_dlq(sqs_client, src_url, dlq_arn):
+    # RedrivePolicy is only parsed via set_queue_attributes, so wire the source here.
     sqs_client.set_queue_attributes(
-        QueueUrl=src,
+        QueueUrl=src_url,
         Attributes={
             "RedrivePolicy": json.dumps(
                 {"deadLetterTargetArn": dlq_arn, "maxReceiveCount": 1}
             )
         },
     )
+
+
+def _redrive_pair(sqs_client, src_name, dlq_name):
+    """Create a source queue wired to a DLQ (maxReceiveCount=1); return (src_url, dlq_url, dlq_arn)."""
+    src = sqs_client.create_queue(QueueName=src_name)["QueueUrl"]
+    dlq = sqs_client.create_queue(QueueName=dlq_name)["QueueUrl"]
+    dlq_arn = _queue_arn(sqs_client, dlq)
+    _wire_to_dlq(sqs_client, src, dlq_arn)
     return src, dlq, dlq_arn
 
 
@@ -222,6 +231,23 @@ def test_list_dead_letter_source_queues(sqs_client):
     assert any("dlq-src" in u for u in urls)
 
 
+def test_list_dead_letter_source_queues_no_sources(sqs_client):
+    # A queue that exists but is not used as a DLQ returns an empty list.
+    url = sqs_client.create_queue(QueueName="dlsq-no-sources")["QueueUrl"]
+    resp = sqs_client.list_dead_letter_source_queues(QueueUrl=url)
+    assert resp["queueUrls"] == []
+
+
+def test_list_dead_letter_source_queues_nonexistent(sqs_client):
+    with pytest.raises(ClientError) as exc:
+        sqs_client.list_dead_letter_source_queues(
+            QueueUrl="http://localhost:4747/000000000000/no-such-dlq"
+        )
+    assert (
+        exc.value.response["Error"]["Code"] == "AWS.SimpleQueueService.NonExistentQueue"
+    )
+
+
 def test_message_move_task_flow(sqs_client):
     src, dlq, dlq_arn = _redrive_pair(sqs_client, "mmt-src", "mmt-dlq")
 
@@ -239,3 +265,66 @@ def test_message_move_task_flow(sqs_client):
 
     cancel = sqs_client.cancel_message_move_task(TaskHandle=handle)
     assert "ApproximateNumberOfMessagesMoved" in cancel
+
+
+def test_message_move_task_destination_routing(sqs_client):
+    # Two sources share a DLQ; an explicit DestinationArn routes deterministically.
+    src1, dlq, dlq_arn = _redrive_pair(sqs_client, "mmt-route-src1", "mmt-route-dlq")
+    src2 = sqs_client.create_queue(QueueName="mmt-route-src2")["QueueUrl"]
+    _wire_to_dlq(sqs_client, src2, dlq_arn)
+    src2_arn = _queue_arn(sqs_client, src2)
+
+    sqs_client.send_message(QueueUrl=dlq, MessageBody="to-src2")
+
+    resp = sqs_client.start_message_move_task(
+        SourceArn=dlq_arn, DestinationArn=src2_arn
+    )
+    assert "TaskHandle" in resp
+
+    msgs = sqs_client.receive_message(QueueUrl=src2, MaxNumberOfMessages=10)
+    assert len(msgs.get("Messages", [])) == 1
+    assert msgs["Messages"][0]["Body"] == "to-src2"
+
+
+def test_message_move_task_ambiguous_destination(sqs_client):
+    # Two sources share a DLQ and no DestinationArn is given -> ambiguous.
+    src1, dlq, dlq_arn = _redrive_pair(sqs_client, "mmt-amb-src1", "mmt-amb-dlq")
+    src2 = sqs_client.create_queue(QueueName="mmt-amb-src2")["QueueUrl"]
+    _wire_to_dlq(sqs_client, src2, dlq_arn)
+
+    with pytest.raises(ClientError) as exc:
+        sqs_client.start_message_move_task(SourceArn=dlq_arn)
+    assert exc.value.response["Error"]["Code"] == "InvalidParameterValue"
+
+
+def test_start_message_move_task_unsupported_source(sqs_client):
+    # SourceArn must be a DLQ; a plain queue is rejected.
+    url = sqs_client.create_queue(QueueName="mmt-not-a-dlq")["QueueUrl"]
+    arn = _queue_arn(sqs_client, url)
+    with pytest.raises(ClientError) as exc:
+        sqs_client.start_message_move_task(SourceArn=arn)
+    assert (
+        exc.value.response["Error"]["Code"]
+        == "AWS.SimpleQueueService.UnsupportedOperation"
+    )
+
+
+def test_start_message_move_task_requires_source_arn(sqs_client):
+    # SourceArn is a required member, so botocore rejects the call client-side.
+    with pytest.raises(ParamValidationError):
+        sqs_client.start_message_move_task(
+            DestinationArn="arn:aws:sqs:us-east-1:000000000000:x"
+        )
+
+
+def test_cancel_message_move_task_missing_handle(sqs_client):
+    # An empty TaskHandle passes client validation and reaches the server handler.
+    with pytest.raises(ClientError) as exc:
+        sqs_client.cancel_message_move_task(TaskHandle="")
+    assert exc.value.response["Error"]["Code"] == "MissingParameter"
+
+
+def test_cancel_message_move_task_unknown_handle(sqs_client):
+    with pytest.raises(ClientError) as exc:
+        sqs_client.cancel_message_move_task(TaskHandle="does-not-exist")
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"

From ce08779ed5fe4f406ee63c812d6ba4bcb8e8dd8f Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Sat, 18 Jul 2026 08:30:29 +0900
Subject: [PATCH 4/6] =?UTF-8?q?fix(sqs):=20address=20review=20=E2=80=94=20?=
 =?UTF-8?q?bounded=20task=20history,=20clearer=20errors,=20more=20tests?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Retain the most recent 10 move-task records per source (was: one per
  source) so ListMessageMoveTasks pagination works while the registry
  stays bounded; clamp MaxResults to [1, 10] like the real API.
- Clarify the sourceQueuesForDLQ lock comment (caller holds s.mu read or
  write) and reword the ambiguous-destination error to describe the
  DLQ/multi-source scenario.
- Add compat tests: ListDeadLetterSourceQueues missing QueueUrl,
  ListMessageMoveTasks required SourceArn and MaxResults semantics
  (default/0/negative -> 10, in-range honored, over-cap clamped).

Required members (SourceArn/QueueUrl) are validated client-side by
botocore (ParamValidationError); the empty-string cases exercise the
server-side MissingParameter path.
---
 internal/services/sqs/provider.go   |  2 +-
 internal/services/sqs/store.go      | 47 ++++++++++++++++++++---------
 internal/services/sqs/store_test.go | 10 +++---
 tests/compatibility/test_sqs.py     | 35 +++++++++++++++++++++
 4 files changed, 75 insertions(+), 19 deletions(-)

diff --git a/internal/services/sqs/provider.go b/internal/services/sqs/provider.go
index d61e2be8..2cf14019 100644
--- a/internal/services/sqs/provider.go
+++ b/internal/services/sqs/provider.go
@@ -645,7 +645,7 @@ func (p *SQSProvider) startMessageMoveTaskJSON(params map[string]any) (*plugin.R
 			"source queue must be configured as a dead-letter queue", http.StatusBadRequest), nil
 	case err == ErrAmbiguousDest:
 		return jsonError("InvalidParameterValue",
-			"DestinationArn is required when the source queue has multiple dead-letter source queues",
+			"DestinationArn is required because the dead-letter queue has more than one source queue",
 			http.StatusBadRequest), nil
 	case err != nil:
 		return nil, err
diff --git a/internal/services/sqs/store.go b/internal/services/sqs/store.go
index 3d1f080a..d035665f 100644
--- a/internal/services/sqs/store.go
+++ b/internal/services/sqs/store.go
@@ -23,7 +23,7 @@ var (
 	ErrReceiptInvalid     = errors.New("receipt handle is invalid")
 	ErrSourceNotDLQ       = errors.New("source queue is not configured as a dead-letter queue")
 	ErrTaskNotFound       = errors.New("message move task not found")
-	ErrAmbiguousDest      = errors.New("destinationArn is required when the dead-letter queue has multiple source queues")
+	ErrAmbiguousDest      = errors.New("DestinationArn is required because the dead-letter queue has more than one source queue")
 )
 
 // QueueInfo holds metadata about an SQS queue.
@@ -677,8 +677,13 @@ type moveTask struct {
 	StartedUnix    int64
 }
 
+// maxTasksPerSource caps how many move-task records are retained per source queue,
+// matching the ListMessageMoveTasks result cap so the registry stays bounded.
+const maxTasksPerSource = 10
+
 // sourceQueuesForDLQ returns the queues under accountID that name dlqName as their
-// dead-letter target. The caller must hold s.mu.
+// dead-letter target. The caller must hold s.mu (read or write); each candidate queue's
+// own mutex is taken briefly to read its redrive target.
 func (s *QueueStore) sourceQueuesForDLQ(dlqName, accountID string) []*queue {
 	var out []*queue
 	for _, q := range s.queues {
@@ -772,14 +777,6 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 	}
 	dst.mu.Unlock()
 
-	// Keep only the latest task per source so the registry stays bounded (a source can have
-	// at most one task record); a new task supersedes any prior one for the same source.
-	for h, t := range s.moveTasks {
-		if t.SourceArn == sourceArn {
-			delete(s.moveTasks, h)
-		}
-	}
-
 	handle := randomID(16)
 	s.moveTasks[handle] = &moveTask{
 		Handle:         handle,
@@ -790,14 +787,36 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 		Moved:          len(moved),
 		StartedUnix:    time.Now().Unix(),
 	}
+	// Retain only the most recent tasks per source so a long-running process that repeatedly
+	// redrives the same source can't grow the registry without limit.
+	s.pruneMoveTasks(sourceArn)
 	return handle, nil
 }
 
-// ListMessageMoveTasks returns move tasks for the given source ARN, newest first,
-// capped at maxResults (default 10 when maxResults <= 0).
+// pruneMoveTasks drops all but the most recent maxTasksPerSource tasks for sourceArn.
+// The caller must hold s.mu (write).
+func (s *QueueStore) pruneMoveTasks(sourceArn string) {
+	var forSource []*moveTask
+	for _, t := range s.moveTasks {
+		if t.SourceArn == sourceArn {
+			forSource = append(forSource, t)
+		}
+	}
+	if len(forSource) <= maxTasksPerSource {
+		return
+	}
+	sort.Slice(forSource, func(i, j int) bool { return forSource[i].StartedUnix > forSource[j].StartedUnix })
+	for _, t := range forSource[maxTasksPerSource:] {
+		delete(s.moveTasks, t.Handle)
+	}
+}
+
+// ListMessageMoveTasks returns move tasks for the given source ARN, newest first. The
+// result count is clamped to [1, maxTasksPerSource]; a non-positive maxResults uses the cap.
 func (s *QueueStore) ListMessageMoveTasks(sourceArn string, maxResults int) []*moveTask {
-	if maxResults <= 0 {
-		maxResults = 10
+	// AWS caps this at 10; a non-positive value defaults to the cap.
+	if maxResults <= 0 || maxResults > maxTasksPerSource {
+		maxResults = maxTasksPerSource
 	}
 
 	s.mu.RLock()
diff --git a/internal/services/sqs/store_test.go b/internal/services/sqs/store_test.go
index 21a33b7a..dfc4b4ea 100644
--- a/internal/services/sqs/store_test.go
+++ b/internal/services/sqs/store_test.go
@@ -287,13 +287,15 @@ func TestQueueStore_MessageMoveTaskRegistryBounded(t *testing.T) {
 		"RedrivePolicy": `{"maxReceiveCount":1,"deadLetterTargetArn":"` + dlqArn + `"}`,
 	}))
 
-	// Starting many tasks for the same source keeps only the latest record.
-	for i := 0; i < 5; i++ {
+	// Starting many tasks for the same source retains only the most recent maxTasksPerSource.
+	for i := 0; i < maxTasksPerSource+5; i++ {
 		_, err := store.StartMessageMoveTask(dlqArn, "", 0, testAccount)
 		require.NoError(t, err)
 	}
-	assert.Len(t, store.moveTasks, 1)
-	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 0), 1)
+	assert.Len(t, store.moveTasks, maxTasksPerSource)
+	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 0), maxTasksPerSource)
+	// A positive MaxResults is honored below the cap.
+	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 3), 3)
 }
 
 func TestQueueStore_Tags(t *testing.T) {
diff --git a/tests/compatibility/test_sqs.py b/tests/compatibility/test_sqs.py
index 3a29d8d1..f4178617 100644
--- a/tests/compatibility/test_sqs.py
+++ b/tests/compatibility/test_sqs.py
@@ -248,6 +248,14 @@ def test_list_dead_letter_source_queues_nonexistent(sqs_client):
     )
 
 
+def test_list_dead_letter_source_queues_missing_queue_url(sqs_client):
+    # QueueUrl is a required member, so omitting it is caught client-side by botocore; an
+    # empty string passes that check and exercises the handler's MissingParameter path.
+    with pytest.raises(ClientError) as exc:
+        sqs_client.list_dead_letter_source_queues(QueueUrl="")
+    assert exc.value.response["Error"]["Code"] == "MissingParameter"
+
+
 def test_message_move_task_flow(sqs_client):
     src, dlq, dlq_arn = _redrive_pair(sqs_client, "mmt-src", "mmt-dlq")
 
@@ -328,3 +336,30 @@ def test_cancel_message_move_task_unknown_handle(sqs_client):
     with pytest.raises(ClientError) as exc:
         sqs_client.cancel_message_move_task(TaskHandle="does-not-exist")
     assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+
+def test_list_message_move_tasks_requires_source_arn(sqs_client):
+    # SourceArn is a required member, so botocore rejects the call client-side.
+    with pytest.raises(ParamValidationError):
+        sqs_client.list_message_move_tasks()
+
+
+def test_list_message_move_tasks_max_results_semantics(sqs_client):
+    _, _, dlq_arn = _redrive_pair(sqs_client, "mmt-max-src", "mmt-max-dlq")
+
+    # Start more tasks than the per-source cap (each start succeeds even with an empty DLQ).
+    for _ in range(15):
+        sqs_client.start_message_move_task(SourceArn=dlq_arn)
+
+    def count(**kw):
+        return len(
+            sqs_client.list_message_move_tasks(SourceArn=dlq_arn, **kw)["Results"]
+        )
+
+    # Default and non-positive values fall back to the 10-task cap.
+    assert count() == 10
+    assert count(MaxResults=0) == 10
+    assert count(MaxResults=-5) == 10
+    # A positive value below the cap is honored; above it is clamped.
+    assert count(MaxResults=7) == 7
+    assert count(MaxResults=25) == 10

From de1b0a27b5ed2c405943cb090f8ecf55fa2e5aad Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Sat, 18 Jul 2026 09:27:56 +0900
Subject: [PATCH 5/6] fix(sqs): address xhigh review of dead-letter redrive
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Redrive only available DLQ messages; leave in-flight (received, undeleted)
  ones so a live consumer's receipt handle stays valid (no duplicate processing)
- ListMessageMoveTasks returns value copies taken under the lock, fixing a
  data race on moveTask.Status vs CancelMessageMoveTask
- Wire the four redrive ops into the Query (form) dispatcher with XML handlers,
  not just the JSON one — Query clients no longer get HTTP 400
- Order move tasks by a monotonic seq (not second-granular StartedUnix), so
  prune/list are deterministic and never drop the just-created task
- CancelMessageMoveTask rejects non-RUNNING tasks (ResourceNotFoundException);
  synchronous redrive is always COMPLETED, matching AWS
- ListMessageMoveTasks MaxResults defaults to 1 (AWS default), not the cap
- Emit StartedTimestamp in epoch milliseconds (AWS Long), not seconds
- Restore docs note that queue-level DelaySeconds/VisibilityTimeout are not enforced
- Hold s.mu only to resolve queue pointers; run drain/fill under per-queue locks
- Extract sortedTasksForSource to dedupe prune/list filter+sort
---
 docs/services/sqs.md                   |   3 +
 internal/services/sqs/provider.go      | 181 +++++++++++++++++++++++--
 internal/services/sqs/provider_test.go |  75 ++++++++++
 internal/services/sqs/store.go         | 106 ++++++++++-----
 internal/services/sqs/store_test.go    |  11 +-
 tests/compatibility/test_sqs.py        |  15 +-
 6 files changed, 333 insertions(+), 58 deletions(-)

diff --git a/docs/services/sqs.md b/docs/services/sqs.md
index 942745ed..aa621c93 100644
--- a/docs/services/sqs.md
+++ b/docs/services/sqs.md
@@ -86,5 +86,8 @@ aws --endpoint-url http://localhost:4747 sqs receive-message \
 - Dead-letter redrive (message move tasks) runs synchronously: tasks complete immediately,
   so there is no `RUNNING` status and `MaxNumberOfMessagesPerSecond` is not throttled
 - `AddPermission`/`RemovePermission` are accepted but not enforced (no access-policy checks)
+- Queue-level `DelaySeconds` and `VisibilityTimeout` attributes are stored and echoed back by
+  `SetQueueAttributes`/`GetQueueAttributes` but not enforced: `SendMessage` ignores `DelaySeconds`
+  and `ReceiveMessage` honors only its per-request `VisibilityTimeout` (default 30s)
 - No message retention/expiry enforcement
 - No long polling (`WaitTimeSeconds` is not honored — receives return immediately)
diff --git a/internal/services/sqs/provider.go b/internal/services/sqs/provider.go
index 2cf14019..842e2b98 100644
--- a/internal/services/sqs/provider.go
+++ b/internal/services/sqs/provider.go
@@ -96,6 +96,15 @@ func (p *SQSProvider) handleQueryRequest(op string, req *http.Request) (*plugin.
 		return p.setQueueAttributes(req)
 	case "PurgeQueue":
 		return p.purgeQueue(req)
+	// Dead-letter redrive
+	case "ListDeadLetterSourceQueues":
+		return p.listDeadLetterSourceQueues(req)
+	case "StartMessageMoveTask":
+		return p.startMessageMoveTask(req)
+	case "ListMessageMoveTasks":
+		return p.listMessageMoveTasks(req)
+	case "CancelMessageMoveTask":
+		return p.cancelMessageMoveTask(req)
 	default:
 		return sqsError("InvalidAction", fmt.Sprintf("unknown action: %s", action), http.StatusBadRequest), nil
 	}
@@ -314,6 +323,44 @@ type purgeQueueResponse struct {
 	XMLName xml.Name `xml:"PurgeQueueResponse"`
 }
 
+type listDeadLetterSourceQueuesResponse struct {
+	XMLName xml.Name `xml:"ListDeadLetterSourceQueuesResponse"`
+	Result  struct {
+		QueueUrls []string `xml:"QueueUrl"`
+	} `xml:"ListDeadLetterSourceQueuesResult"`
+}
+
+type startMessageMoveTaskResponse struct {
+	XMLName xml.Name `xml:"StartMessageMoveTaskResponse"`
+	Result  struct {
+		TaskHandle string `xml:"TaskHandle"`
+	} `xml:"StartMessageMoveTaskResult"`
+}
+
+type listMessageMoveTasksResponse struct {
+	XMLName xml.Name `xml:"ListMessageMoveTasksResponse"`
+	Result  struct {
+		Results []moveTaskXML `xml:"ListMessageMoveTasksResultEntry"`
+	} `xml:"ListMessageMoveTasksResult"`
+}
+
+type moveTaskXML struct {
+	TaskHandle                       string `xml:"TaskHandle"`
+	Status                           string `xml:"Status"`
+	SourceArn                        string `xml:"SourceArn"`
+	DestinationArn                   string `xml:"DestinationArn,omitempty"`
+	MaxNumberOfMessagesPerSecond     int    `xml:"MaxNumberOfMessagesPerSecond"`
+	ApproximateNumberOfMessagesMoved int    `xml:"ApproximateNumberOfMessagesMoved"`
+	StartedTimestamp                 int64  `xml:"StartedTimestamp"`
+}
+
+type cancelMessageMoveTaskResponse struct {
+	XMLName xml.Name `xml:"CancelMessageMoveTaskResponse"`
+	Result  struct {
+		ApproximateNumberOfMessagesMoved int `xml:"ApproximateNumberOfMessagesMoved"`
+	} `xml:"CancelMessageMoveTaskResult"`
+}
+
 type messageAttributeXML struct {
 	Name  string                   `xml:"Name"`
 	Value messageAttributeValueXML `xml:"Value"`
@@ -637,18 +684,12 @@ func (p *SQSProvider) startMessageMoveTaskJSON(params map[string]any) (*plugin.R
 	maxRate := intParam(params, "MaxNumberOfMessagesPerSecond", 0)
 
 	handle, err := p.store.StartMessageMoveTask(sourceArn, destArn, maxRate, defaultAccountID)
-	switch {
-	case err == ErrQueueNotFound:
-		return jsonError("ResourceNotFoundException", "queue not found", http.StatusBadRequest), nil
-	case err == ErrSourceNotDLQ:
-		return jsonError("AWS.SimpleQueueService.UnsupportedOperation",
-			"source queue must be configured as a dead-letter queue", http.StatusBadRequest), nil
-	case err == ErrAmbiguousDest:
-		return jsonError("InvalidParameterValue",
-			"DestinationArn is required because the dead-letter queue has more than one source queue",
-			http.StatusBadRequest), nil
-	case err != nil:
-		return nil, err
+	if err != nil {
+		code, msg := startMoveTaskError(err)
+		if code == "" {
+			return nil, err
+		}
+		return jsonError(code, msg, http.StatusBadRequest), nil
 	}
 
 	return jsonResp(http.StatusOK, map[string]any{"TaskHandle": handle})
@@ -670,7 +711,7 @@ func (p *SQSProvider) listMessageMoveTasksJSON(params map[string]any) (*plugin.R
 			"SourceArn":                        t.SourceArn,
 			"MaxNumberOfMessagesPerSecond":     t.MaxRate,
 			"ApproximateNumberOfMessagesMoved": t.Moved,
-			"StartedTimestamp":                 t.StartedUnix,
+			"StartedTimestamp":                 t.StartedMillis,
 		}
 		if t.DestinationArn != "" {
 			entry["DestinationArn"] = t.DestinationArn
@@ -688,8 +729,14 @@ func (p *SQSProvider) cancelMessageMoveTaskJSON(params map[string]any) (*plugin.
 	}
 
 	moved, err := p.store.CancelMessageMoveTask(handle)
-	if err != nil {
+	switch {
+	case err == ErrTaskNotFound:
 		return jsonError("ResourceNotFoundException", "task not found", http.StatusBadRequest), nil
+	case err == ErrTaskNotCancellable:
+		// AWS only cancels RUNNING tasks; a completed task is reported as not found.
+		return jsonError("ResourceNotFoundException", "task is not running and cannot be cancelled", http.StatusBadRequest), nil
+	case err != nil:
+		return nil, err
 	}
 
 	return jsonResp(http.StatusOK, map[string]any{"ApproximateNumberOfMessagesMoved": moved})
@@ -981,6 +1028,112 @@ func (p *SQSProvider) purgeQueue(req *http.Request) (*plugin.Response, error) {
 	return xmlResp(http.StatusOK, purgeQueueResponse{})
 }
 
+// startMoveTaskError maps a StartMessageMoveTask store error to its AWS error code and
+// message. It returns an empty code for errors that should surface as a 500 (nil err too).
+func startMoveTaskError(err error) (code, message string) {
+	switch err {
+	case ErrQueueNotFound:
+		return "ResourceNotFoundException", "queue not found"
+	case ErrSourceNotDLQ:
+		return "AWS.SimpleQueueService.UnsupportedOperation", "source queue must be configured as a dead-letter queue"
+	case ErrAmbiguousDest:
+		return "InvalidParameterValue", "DestinationArn is required because the dead-letter queue has more than one source queue"
+	default:
+		return "", ""
+	}
+}
+
+// --- Dead-letter redrive (Query protocol) ---
+
+func (p *SQSProvider) listDeadLetterSourceQueues(req *http.Request) (*plugin.Response, error) {
+	queueURL := req.FormValue("QueueUrl")
+	if queueURL == "" {
+		return sqsError("MissingParameter", "QueueUrl is required", http.StatusBadRequest), nil
+	}
+
+	name := queueNameFromURL(queueURL)
+	urls, err := p.store.ListDeadLetterSourceQueues(name, defaultAccountID)
+	if err != nil {
+		return sqsError("AWS.SimpleQueueService.NonExistentQueue", "queue not found", http.StatusBadRequest), nil
+	}
+
+	resp := listDeadLetterSourceQueuesResponse{}
+	resp.Result.QueueUrls = urls
+	return xmlResp(http.StatusOK, resp)
+}
+
+func (p *SQSProvider) startMessageMoveTask(req *http.Request) (*plugin.Response, error) {
+	sourceArn := req.FormValue("SourceArn")
+	if sourceArn == "" {
+		return sqsError("MissingParameter", "SourceArn is required", http.StatusBadRequest), nil
+	}
+	destArn := req.FormValue("DestinationArn")
+	maxRate := 0
+	if s := req.FormValue("MaxNumberOfMessagesPerSecond"); s != "" {
+		maxRate, _ = strconv.Atoi(s)
+	}
+
+	handle, err := p.store.StartMessageMoveTask(sourceArn, destArn, maxRate, defaultAccountID)
+	if err != nil {
+		code, msg := startMoveTaskError(err)
+		if code == "" {
+			return nil, err
+		}
+		return sqsError(code, msg, http.StatusBadRequest), nil
+	}
+
+	resp := startMessageMoveTaskResponse{}
+	resp.Result.TaskHandle = handle
+	return xmlResp(http.StatusOK, resp)
+}
+
+func (p *SQSProvider) listMessageMoveTasks(req *http.Request) (*plugin.Response, error) {
+	sourceArn := req.FormValue("SourceArn")
+	if sourceArn == "" {
+		return sqsError("MissingParameter", "SourceArn is required", http.StatusBadRequest), nil
+	}
+	maxResults := 0
+	if s := req.FormValue("MaxResults"); s != "" {
+		maxResults, _ = strconv.Atoi(s)
+	}
+
+	tasks := p.store.ListMessageMoveTasks(sourceArn, maxResults)
+	resp := listMessageMoveTasksResponse{}
+	for _, t := range tasks {
+		resp.Result.Results = append(resp.Result.Results, moveTaskXML{
+			TaskHandle:                       t.Handle,
+			Status:                           t.Status,
+			SourceArn:                        t.SourceArn,
+			DestinationArn:                   t.DestinationArn,
+			MaxNumberOfMessagesPerSecond:     t.MaxRate,
+			ApproximateNumberOfMessagesMoved: t.Moved,
+			StartedTimestamp:                 t.StartedMillis,
+		})
+	}
+	return xmlResp(http.StatusOK, resp)
+}
+
+func (p *SQSProvider) cancelMessageMoveTask(req *http.Request) (*plugin.Response, error) {
+	handle := req.FormValue("TaskHandle")
+	if handle == "" {
+		return sqsError("MissingParameter", "TaskHandle is required", http.StatusBadRequest), nil
+	}
+
+	moved, err := p.store.CancelMessageMoveTask(handle)
+	switch {
+	case err == ErrTaskNotFound:
+		return sqsError("ResourceNotFoundException", "task not found", http.StatusBadRequest), nil
+	case err == ErrTaskNotCancellable:
+		return sqsError("ResourceNotFoundException", "task is not running and cannot be cancelled", http.StatusBadRequest), nil
+	case err != nil:
+		return nil, err
+	}
+
+	resp := cancelMessageMoveTaskResponse{}
+	resp.Result.ApproximateNumberOfMessagesMoved = moved
+	return xmlResp(http.StatusOK, resp)
+}
+
 // parseFormMessageAttributes extracts MessageAttribute.N.* from form values.
 func parseFormMessageAttributes(req *http.Request) map[string]MessageAttribute {
 	attrs := make(map[string]MessageAttribute)
diff --git a/internal/services/sqs/provider_test.go b/internal/services/sqs/provider_test.go
index 3f40b196..f07ff39a 100644
--- a/internal/services/sqs/provider_test.go
+++ b/internal/services/sqs/provider_test.go
@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http/httptest"
+	"net/url"
 	"strings"
 	"testing"
 
@@ -464,3 +465,77 @@ func TestSQSProvider_Init_PortFromOptions(t *testing.T) {
 		})
 	}
 }
+
+// TestSQSProvider_Query_Redrive exercises the four dead-letter redrive ops over the legacy
+// Query (form-encoded) protocol, which the JSON dispatcher's siblings must not be the only
+// path for. Regression guard: these ops were once wired only into the JSON dispatcher, so
+// Query clients got HTTP 400.
+func TestSQSProvider_Query_Redrive(t *testing.T) {
+	p := newTestSQSProvider(t)
+
+	const acct = defaultAccountID
+	dlqArn := "arn:aws:sqs:us-east-1:" + acct + ":redrive-dlq"
+	dlqURL := fmt.Sprintf("http://localhost:4747/%s/redrive-dlq", acct)
+	srcURL := fmt.Sprintf("http://localhost:4747/%s/redrive-src", acct)
+
+	handleForm(t, p, "Action=CreateQueue&QueueName=redrive-src")
+	handleForm(t, p, "Action=CreateQueue&QueueName=redrive-dlq")
+
+	// Wire the source to the DLQ so redrive is legal.
+	setAttrs := url.Values{
+		"Action":            {"SetQueueAttributes"},
+		"QueueUrl":          {srcURL},
+		"Attribute.1.Name":  {"RedrivePolicy"},
+		"Attribute.1.Value": {`{"maxReceiveCount":1,"deadLetterTargetArn":"` + dlqArn + `"}`},
+	}
+	require.Equal(t, 200, handleForm(t, p, setAttrs.Encode()).StatusCode)
+
+	// ListDeadLetterSourceQueues returns the source URL.
+	dl := handleForm(t, p, (url.Values{"Action": {"ListDeadLetterSourceQueues"}, "QueueUrl": {dlqURL}}).Encode())
+	assert.Equal(t, 200, dl.StatusCode)
+	assert.Contains(t, string(dl.Body), "ListDeadLetterSourceQueuesResponse")
+	assert.Contains(t, string(dl.Body), "redrive-src")
+
+	// Put a message in the DLQ, then redrive it back to the source.
+	handleForm(t, p, (url.Values{"Action": {"SendMessage"}, "QueueUrl": {dlqURL}, "MessageBody": {"orphan"}}).Encode())
+
+	start := handleForm(t, p, (url.Values{"Action": {"StartMessageMoveTask"}, "SourceArn": {dlqArn}}).Encode())
+	require.Equal(t, 200, start.StatusCode)
+	assert.Contains(t, string(start.Body), "StartMessageMoveTaskResponse")
+	handle := xmlField(t, start.Body, "TaskHandle")
+	require.NotEmpty(t, handle)
+
+	// ListMessageMoveTasks reports the completed task.
+	list := handleForm(t, p, (url.Values{"Action": {"ListMessageMoveTasks"}, "SourceArn": {dlqArn}}).Encode())
+	assert.Equal(t, 200, list.StatusCode)
+	assert.Contains(t, string(list.Body), "COMPLETED")
+	assert.Contains(t, string(list.Body), "ListMessageMoveTasksResultEntry")
+
+	// Cancel is rejected because the synchronous redrive already completed.
+	cancel := handleForm(t, p, (url.Values{"Action": {"CancelMessageMoveTask"}, "TaskHandle": {handle}}).Encode())
+	assert.Equal(t, 400, cancel.StatusCode)
+	assert.Contains(t, string(cancel.Body), "ResourceNotFoundException")
+
+	// A plain queue (not a DLQ) cannot start a move task.
+	srcArn := "arn:aws:sqs:us-east-1:" + acct + ":redrive-src"
+	bad := handleForm(t, p, (url.Values{"Action": {"StartMessageMoveTask"}, "SourceArn": {srcArn}}).Encode())
+	assert.Equal(t, 400, bad.StatusCode)
+	assert.Contains(t, string(bad.Body), "UnsupportedOperation")
+}
+
+// xmlField extracts the text of the first <tag>…</tag> element from an XML body.
+func xmlField(t *testing.T, body []byte, tag string) string {
+	t.Helper()
+	open, close := "<"+tag+">", "</"+tag+">"
+	s := string(body)
+	i := strings.Index(s, open)
+	if i < 0 {
+		return ""
+	}
+	i += len(open)
+	j := strings.Index(s[i:], close)
+	if j < 0 {
+		return ""
+	}
+	return s[i : i+j]
+}
diff --git a/internal/services/sqs/store.go b/internal/services/sqs/store.go
index d035665f..3e07e5cc 100644
--- a/internal/services/sqs/store.go
+++ b/internal/services/sqs/store.go
@@ -23,6 +23,7 @@ var (
 	ErrReceiptInvalid     = errors.New("receipt handle is invalid")
 	ErrSourceNotDLQ       = errors.New("source queue is not configured as a dead-letter queue")
 	ErrTaskNotFound       = errors.New("message move task not found")
+	ErrTaskNotCancellable = errors.New("message move task is not running and cannot be cancelled")
 	ErrAmbiguousDest      = errors.New("DestinationArn is required because the dead-letter queue has more than one source queue")
 )
 
@@ -86,10 +87,11 @@ type queue struct {
 
 // QueueStore is a thread-safe in-memory store for SQS queues.
 type QueueStore struct {
-	mu        sync.RWMutex
-	queues    map[string]*queue    // key: "accountID/queueName"
-	moveTasks map[string]*moveTask // key: task handle
-	port      int                  // HTTP port for queue-URL construction
+	mu          sync.RWMutex
+	queues      map[string]*queue    // key: "accountID/queueName"
+	moveTasks   map[string]*moveTask // key: task handle
+	moveTaskSeq int64                // monotonic sequence for deterministic move-task ordering
+	port        int                  // HTTP port for queue-URL construction
 }
 
 // NewQueueStore creates and returns an empty QueueStore bound to the given
@@ -666,15 +668,16 @@ func (s *QueueStore) ListQueueTags(queueName, accountID string) (map[string]stri
 // --- Dead-letter redrive (message move tasks) ---
 
 // moveTask records a dead-letter redrive task. Redrive runs synchronously (in-memory),
-// so a task is created already in a terminal state.
+// so a task is created already in the terminal COMPLETED state (there is no RUNNING phase).
 type moveTask struct {
 	Handle         string
 	SourceArn      string
 	DestinationArn string
 	MaxRate        int    // MaxNumberOfMessagesPerSecond; 0 = unlimited
-	Status         string // "COMPLETED" | "CANCELLED"
+	Status         string // always "COMPLETED" (synchronous redrive)
 	Moved          int
-	StartedUnix    int64
+	StartedMillis  int64 // epoch milliseconds, matching AWS's StartedTimestamp (a Long)
+	seq            int64 // monotonic insertion order; sole ordering key (StartedMillis can collide)
 }
 
 // maxTasksPerSource caps how many move-task records are retained per source queue,
@@ -724,16 +727,19 @@ func (s *QueueStore) ListDeadLetterSourceQueues(dlqName, accountID string) ([]st
 func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int, accountID string) (string, error) {
 	srcName := arnToQueueName(sourceArn)
 
+	// Resolve the source and destination queue pointers under s.mu, then release it so the
+	// drain/fill runs under only the per-queue locks — the store-wide lock must not be held
+	// across a potentially large message move (it would block all other queues).
 	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	src, ok := s.queues[queueKey(accountID, srcName)]
 	if !ok {
+		s.mu.Unlock()
 		return "", ErrQueueNotFound
 	}
 
 	sources := s.sourceQueuesForDLQ(srcName, accountID)
 	if len(sources) == 0 {
+		s.mu.Unlock()
 		return "", ErrSourceNotDLQ
 	}
 
@@ -741,6 +747,7 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 	if destArn != "" {
 		dst, ok = s.queues[queueKey(accountID, arnToQueueName(destArn))]
 		if !ok {
+			s.mu.Unlock()
 			return "", ErrQueueNotFound
 		}
 	} else {
@@ -748,23 +755,34 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 		// origin isn't tracked, so this is only unambiguous when the DLQ has a single source;
 		// with multiple sources the caller must pass DestinationArn.
 		if len(sources) > 1 {
+			s.mu.Unlock()
 			return "", ErrAmbiguousDest
 		}
 		dst = sources[0]
 	}
+	s.mu.Unlock()
 
 	// ponytail: redrive runs synchronously with no RUNNING state or rate limiting;
 	// upgrade to a goroutine + MaxRate throttle if timing fidelity is ever needed.
 	// Lock only one queue at a time (drain the source, then fill the destination) to stay
 	// deadlock-free with the source→DLQ path in ReceiveMessage.
+	now := time.Now()
 	src.mu.Lock()
 	moved := make([]*Message, 0, len(src.messages))
+	remaining := make([]*Message, 0)
 	for _, m := range src.messages {
-		if !m.deleted {
-			moved = append(moved, m)
+		if m.deleted {
+			continue // compacted away, as before
 		}
+		if now.Before(m.invisibleUntil) {
+			// In-flight (received but not yet deleted): leave it in the DLQ so the consumer's
+			// receipt handle stays valid. Only available messages are redriven.
+			remaining = append(remaining, m)
+			continue
+		}
+		moved = append(moved, m)
 	}
-	src.messages = make([]*Message, 0)
+	src.messages = remaining
 	src.mu.Unlock()
 
 	dst.mu.Lock()
@@ -778,6 +796,8 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 	dst.mu.Unlock()
 
 	handle := randomID(16)
+	s.mu.Lock()
+	s.moveTaskSeq++
 	s.moveTasks[handle] = &moveTask{
 		Handle:         handle,
 		SourceArn:      sourceArn,
@@ -785,58 +805,73 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int
 		MaxRate:        maxRate,
 		Status:         "COMPLETED",
 		Moved:          len(moved),
-		StartedUnix:    time.Now().Unix(),
+		StartedMillis:  now.UnixMilli(),
+		seq:            s.moveTaskSeq,
 	}
 	// Retain only the most recent tasks per source so a long-running process that repeatedly
 	// redrives the same source can't grow the registry without limit.
 	s.pruneMoveTasks(sourceArn)
+	s.mu.Unlock()
 	return handle, nil
 }
 
-// pruneMoveTasks drops all but the most recent maxTasksPerSource tasks for sourceArn.
-// The caller must hold s.mu (write).
-func (s *QueueStore) pruneMoveTasks(sourceArn string) {
-	var forSource []*moveTask
+// sortedTasksForSource returns the move tasks for sourceArn, newest first (by insertion
+// sequence, which is unique and monotonic — StartedMillis alone can collide). The caller
+// must hold s.mu (read or write).
+func (s *QueueStore) sortedTasksForSource(sourceArn string) []*moveTask {
+	var out []*moveTask
 	for _, t := range s.moveTasks {
 		if t.SourceArn == sourceArn {
-			forSource = append(forSource, t)
+			out = append(out, t)
 		}
 	}
+	sort.Slice(out, func(i, j int) bool { return out[i].seq > out[j].seq })
+	return out
+}
+
+// pruneMoveTasks drops all but the most recent maxTasksPerSource tasks for sourceArn.
+// The caller must hold s.mu (write).
+func (s *QueueStore) pruneMoveTasks(sourceArn string) {
+	forSource := s.sortedTasksForSource(sourceArn)
 	if len(forSource) <= maxTasksPerSource {
 		return
 	}
-	sort.Slice(forSource, func(i, j int) bool { return forSource[i].StartedUnix > forSource[j].StartedUnix })
 	for _, t := range forSource[maxTasksPerSource:] {
 		delete(s.moveTasks, t.Handle)
 	}
 }
 
-// ListMessageMoveTasks returns move tasks for the given source ARN, newest first. The
-// result count is clamped to [1, maxTasksPerSource]; a non-positive maxResults uses the cap.
-func (s *QueueStore) ListMessageMoveTasks(sourceArn string, maxResults int) []*moveTask {
-	// AWS caps this at 10; a non-positive value defaults to the cap.
-	if maxResults <= 0 || maxResults > maxTasksPerSource {
+// ListMessageMoveTasks returns copies of the move tasks for the given source ARN, newest
+// first. Copies (not shared pointers) are returned so callers can read the fields without
+// holding s.mu while CancelMessageMoveTask may mutate the live records. The result count is
+// clamped to [1, maxTasksPerSource]; a non-positive maxResults defaults to 1 (AWS's default).
+func (s *QueueStore) ListMessageMoveTasks(sourceArn string, maxResults int) []moveTask {
+	if maxResults <= 0 {
+		maxResults = 1
+	}
+	if maxResults > maxTasksPerSource {
 		maxResults = maxTasksPerSource
 	}
 
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
-	var out []*moveTask
-	for _, t := range s.moveTasks {
-		if t.SourceArn == sourceArn {
-			out = append(out, t)
-		}
+	forSource := s.sortedTasksForSource(sourceArn)
+	if len(forSource) > maxResults {
+		forSource = forSource[:maxResults]
 	}
-	sort.Slice(out, func(i, j int) bool { return out[i].StartedUnix > out[j].StartedUnix })
-	if len(out) > maxResults {
-		out = out[:maxResults]
+	out := make([]moveTask, len(forSource))
+	for i, t := range forSource {
+		out[i] = *t
 	}
 	return out
 }
 
-// CancelMessageMoveTask cancels the task with the given handle and returns the number of
-// messages already moved. Returns ErrTaskNotFound if the handle is unknown.
+// CancelMessageMoveTask cancels the running task with the given handle and returns the number
+// of messages already moved. Returns ErrTaskNotFound if the handle is unknown, or
+// ErrTaskNotCancellable if the task is not RUNNING. Because redrive here is synchronous, tasks
+// are always already COMPLETED, so — matching AWS, which only cancels RUNNING tasks — a cancel
+// never succeeds.
 func (s *QueueStore) CancelMessageMoveTask(handle string) (int, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -845,6 +880,9 @@ func (s *QueueStore) CancelMessageMoveTask(handle string) (int, error) {
 	if !ok {
 		return 0, ErrTaskNotFound
 	}
+	if t.Status != "RUNNING" {
+		return t.Moved, ErrTaskNotCancellable
+	}
 	t.Status = "CANCELLED"
 	return t.Moved, nil
 }
diff --git a/internal/services/sqs/store_test.go b/internal/services/sqs/store_test.go
index dfc4b4ea..39be7a9c 100644
--- a/internal/services/sqs/store_test.go
+++ b/internal/services/sqs/store_test.go
@@ -232,9 +232,10 @@ func TestQueueStore_MessageMoveTaskLifecycle(t *testing.T) {
 	require.Len(t, back, 1)
 	assert.Equal(t, "redrive-me", back[0].Body)
 
-	// Cancel reports the moved count.
+	// Cancel is rejected: the synchronous redrive already COMPLETED, so there is no RUNNING
+	// task to cancel (matching AWS, which only cancels running tasks). It still reports the count.
 	moved, err := store.CancelMessageMoveTask(handle)
-	require.NoError(t, err)
+	assert.ErrorIs(t, err, ErrTaskNotCancellable)
 	assert.Equal(t, 1, moved)
 
 	// An unknown handle errors.
@@ -293,9 +294,11 @@ func TestQueueStore_MessageMoveTaskRegistryBounded(t *testing.T) {
 		require.NoError(t, err)
 	}
 	assert.Len(t, store.moveTasks, maxTasksPerSource)
-	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 0), maxTasksPerSource)
-	// A positive MaxResults is honored below the cap.
+	// A non-positive MaxResults defaults to 1 (AWS: the most recent task).
+	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 0), 1)
+	// A positive MaxResults is honored below the cap, and clamped to it above.
 	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 3), 3)
+	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 25), maxTasksPerSource)
 }
 
 func TestQueueStore_Tags(t *testing.T) {
diff --git a/tests/compatibility/test_sqs.py b/tests/compatibility/test_sqs.py
index f4178617..5e8693cf 100644
--- a/tests/compatibility/test_sqs.py
+++ b/tests/compatibility/test_sqs.py
@@ -271,8 +271,11 @@ def test_message_move_task_flow(sqs_client):
     assert len(tasks) >= 1
     assert tasks[0]["SourceArn"] == dlq_arn
 
-    cancel = sqs_client.cancel_message_move_task(TaskHandle=handle)
-    assert "ApproximateNumberOfMessagesMoved" in cancel
+    # The synchronous redrive already completed, so there is no RUNNING task to cancel
+    # (AWS only cancels running tasks).
+    with pytest.raises(ClientError) as exc:
+        sqs_client.cancel_message_move_task(TaskHandle=handle)
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
 
 
 def test_message_move_task_destination_routing(sqs_client):
@@ -356,10 +359,10 @@ def count(**kw):
             sqs_client.list_message_move_tasks(SourceArn=dlq_arn, **kw)["Results"]
         )
 
-    # Default and non-positive values fall back to the 10-task cap.
-    assert count() == 10
-    assert count(MaxResults=0) == 10
-    assert count(MaxResults=-5) == 10
+    # Default and non-positive values fall back to AWS's documented default of 1.
+    assert count() == 1
+    assert count(MaxResults=0) == 1
+    assert count(MaxResults=-5) == 1
     # A positive value below the cap is honored; above it is clamped.
     assert count(MaxResults=7) == 7
     assert count(MaxResults=25) == 10

From abb7659a2f306f81053098f5771a94f0dfaede22 Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Sat, 18 Jul 2026 10:28:04 +0900
Subject: [PATCH 6/6] test(sqs): assert move-task pruning keeps newest and
 preserves seq order
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Strengthen TestQueueStore_MessageMoveTaskRegistryBounded to verify which
tasks survive pruning (oldest removed, newest maxTasksPerSource kept) and
that ListMessageMoveTasks returns them newest-first — guarding the
monotonic seq ordering when tasks share a StartedMillis.
---
 internal/services/sqs/store_test.go | 31 +++++++++++++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/internal/services/sqs/store_test.go b/internal/services/sqs/store_test.go
index 39be7a9c..bc437b3a 100644
--- a/internal/services/sqs/store_test.go
+++ b/internal/services/sqs/store_test.go
@@ -289,13 +289,40 @@ func TestQueueStore_MessageMoveTaskRegistryBounded(t *testing.T) {
 	}))
 
 	// Starting many tasks for the same source retains only the most recent maxTasksPerSource.
-	for i := 0; i < maxTasksPerSource+5; i++ {
-		_, err := store.StartMessageMoveTask(dlqArn, "", 0, testAccount)
+	// Capture the handles in insertion order so we can assert which ones survive pruning.
+	const extra = 5
+	handles := make([]string, 0, maxTasksPerSource+extra)
+	for i := 0; i < maxTasksPerSource+extra; i++ {
+		h, err := store.StartMessageMoveTask(dlqArn, "", 0, testAccount)
 		require.NoError(t, err)
+		handles = append(handles, h)
 	}
 	assert.Len(t, store.moveTasks, maxTasksPerSource)
+
+	// The oldest `extra` tasks are pruned; the newest maxTasksPerSource are kept.
+	for _, h := range handles[:extra] {
+		assert.NotContains(t, store.moveTasks, h, "oldest tasks should be pruned")
+	}
+	for _, h := range handles[extra:] {
+		assert.Contains(t, store.moveTasks, h, "newest tasks should be retained")
+	}
+
+	// ListMessageMoveTasks returns the kept tasks newest-first (reverse insertion order),
+	// even though they share a StartedMillis — ordering relies on the monotonic seq.
+	got := store.ListMessageMoveTasks(dlqArn, maxTasksPerSource)
+	gotHandles := make([]string, len(got))
+	for i, tk := range got {
+		gotHandles[i] = tk.Handle
+	}
+	want := make([]string, maxTasksPerSource)
+	for i, h := range handles[extra:] {
+		want[len(want)-1-i] = h
+	}
+	assert.Equal(t, want, gotHandles)
+
 	// A non-positive MaxResults defaults to 1 (AWS: the most recent task).
 	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 0), 1)
+	assert.Equal(t, handles[len(handles)-1], store.ListMessageMoveTasks(dlqArn, 0)[0].Handle)
 	// A positive MaxResults is honored below the cap, and clamped to it above.
 	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 3), 3)
 	assert.Len(t, store.ListMessageMoveTasks(dlqArn, 25), maxTasksPerSource)