fix(sqs): reject FIFO->Standard redrive policies, not just Standard->FIFO

The existing guard in redriveCandidateToDLQ only rejected the
Standard-source -> FIFO-DLQ direction. The symmetric mismatch
(FIFO source -> Standard DLQ) still committed: buildDLQRecord
copies srcRec.MessageGroupId verbatim, and tryDeliverCandidate
takes its FIFO group-lock path whenever MessageGroupId != "" -
serializing delivery in a queue that should behave as Standard.

Fix: load srcMeta in addition to dlqMeta and require
srcMeta.IsFIFO == dlqMeta.IsFIFO before building the OCC dispatch.
This is the runtime defense AWS relies on - the catalog accepts a
RedrivePolicy that names a queue created or recreated as a different
type later, so attribute-time validation cannot catch every case.

The existing FIFO-DLQ + empty-MessageGroupId guard is kept as a
defense-in-depth rail for malformed FIFO->FIFO records that slip
past the type-equality check. Validation logic moved to a
validateRedriveTargets helper so redriveCandidateToDLQ stays under
the cyclop=10 ceiling per project rules - no //nolint.

Adds TestSQSServer_RedrivePolicyStandardDlqRejectsFifoSource which
fails on the prior code (DLQ ends up with a record carrying
MessageGroupId:g1) and passes after the fix.

Author: bootjp

adapter/sqs_extra_test.go

@@ -1654,6 +1654,77 @@ func TestSQSServer_RedrivePolicyFifoDlqRejectsStandardSource(t *testing.T) {
 	}
 }
 
+func TestSQSServer_RedrivePolicyStandardDlqRejectsFifoSource(t *testing.T) {
+	t.Parallel()
+	// Symmetric to RedrivePolicyFifoDlqRejectsStandardSource: the
+	// existing guard rejected only the Standard→FIFO direction, but
+	// FIFO→Standard was equally broken. A FIFO source carries
+	// MessageGroupId on every record; copying that into a Standard
+	// DLQ and then issuing ReceiveMessage on the DLQ trips
+	// tryDeliverCandidate's FIFO group-lock branch (gated solely on
+	// MessageGroupId != ""), which serializes delivery in a queue
+	// clients believe behaves as Standard. AWS rejects mixed-type
+	// redrive policies; this test pins that behavior.
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	// Standard DLQ (no FifoQueue attribute).
+	_, out := callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName": "dlq-standard",
+	})
+	dlqURL, _ := out["QueueUrl"].(string)
+	if dlqURL == "" {
+		t.Fatalf("Standard DLQ create failed: %v", out)
+	}
+
+	// FIFO source with RedrivePolicy targeting the Standard DLQ.
+	policy := `{"deadLetterTargetArn":"arn:aws:sqs:us-east-1:000000000000:dlq-standard","maxReceiveCount":1}`
+	status, out := callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName": "src-fifo.fifo",
+		"Attributes": map[string]string{
+			"FifoQueue":     "true",
+			"RedrivePolicy": policy,
+		},
+	})
+	if status != http.StatusOK {
+		t.Fatalf("create FIFO source with Standard-DLQ policy: %d %v", status, out)
+	}
+	srcURL, _ := out["QueueUrl"].(string)
+
+	_, _ = callSQS(t, node, sqsSendMessageTarget, map[string]any{
+		"QueueUrl":               srcURL,
+		"MessageBody":            "poison",
+		"MessageGroupId":         "g1",
+		"MessageDeduplicationId": "d1",
+	})
+	// First receive bumps ReceiveCount to 1 == maxReceiveCount;
+	// second receive (after visibility expiry) should attempt redrive.
+	// The runtime type-compatibility gate must trip, leaving the DLQ
+	// empty and the source message intact.
+	_, _ = callSQS(t, node, sqsReceiveMessageTarget, map[string]any{
+		"QueueUrl":            srcURL,
+		"MaxNumberOfMessages": 1,
+		"VisibilityTimeout":   1,
+	})
+	time.Sleep(1100 * time.Millisecond)
+	_, _ = callSQS(t, node, sqsReceiveMessageTarget, map[string]any{
+		"QueueUrl":            srcURL,
+		"MaxNumberOfMessages": 1,
+	})
+
+	// Direct DLQ receive: a successful redrive would have written a
+	// record with non-empty MessageGroupId here; the fix must keep
+	// the DLQ empty.
+	_, body := callSQS(t, node, sqsReceiveMessageTarget, map[string]any{
+		"QueueUrl":            dlqURL,
+		"MaxNumberOfMessages": 10,
+	})
+	if msgs, _ := body["Messages"].([]any); len(msgs) > 0 {
+		t.Fatalf("Standard DLQ received a redriven FIFO record (would carry MessageGroupId): %v", msgs)
+	}
+}
+
 func TestSQSServer_TagQueueRequiresTags(t *testing.T) {
 	t.Parallel()
 	nodes, _, _ := createNode(t, 1)

adapter/sqs_redrive.go

@@ -121,38 +121,9 @@ func (s *SQSServer) redriveCandidateToDLQ(
 	srcArn string,
 	readTS uint64,
 ) (bool, error) {
-	// Self-referential RedrivePolicy is rejected at attribute-apply
-	// time, but a defense-in-depth check here keeps the receive path
-	// safe even against records committed before the validator
-	// existed (or under a future relaxation of the validator).
-	// Without this, redrive would delete and rewrite the same record
-	// in place, looping the poison message forever.
-	if policy.DLQName == srcQueueName {
-		return false, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
-			"RedrivePolicy.deadLetterTargetArn must not point at the source queue")
-	}
-	dlqMeta, exists, err := s.loadQueueMetaAt(ctx, policy.DLQName, readTS)
+	dlqMeta, err := s.validateRedriveTargets(ctx, srcQueueName, srcRec, policy, readTS)
 	if err != nil {
-		return false, errors.WithStack(err)
-	}
-	if !exists {
-		// DLQ vanished between policy-set and receive. Refusing the
-		// move keeps the source message in flight; the operator can
-		// recreate the DLQ or detach the policy.
-		return false, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
-			"RedrivePolicy targets non-existent DLQ "+policy.DLQName)
-	}
-	// FIFO DLQs require source records to carry MessageGroupId so the
-	// DLQ-side ReceiveMessage enforces group-lock semantics. Without
-	// this gate, a redrive from a Standard source into a FIFO DLQ
-	// would write records with empty MessageGroupId — tryDeliverCandidate
-	// only takes the FIFO path when MessageGroupId is non-empty, so
-	// those messages would bypass FIFO ordering inside what clients
-	// believe is a strictly-ordered queue. We refuse the move and
-	// surface the misconfiguration to operators.
-	if dlqMeta.IsFIFO && srcRec.MessageGroupId == "" {
-		return false, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
-			"FIFO DLQ requires source records to carry MessageGroupId")
+		return false, err
 	}
 	dlqRec, dlqRecordBytes, err := buildDLQRecord(srcRec, dlqMeta, srcArn)
 	if err != nil {
@@ -171,6 +142,61 @@ func (s *SQSServer) redriveCandidateToDLQ(
 	return true, nil
 }
 
+// validateRedriveTargets enforces every static precondition on the
+// (source, DLQ, policy) triple before the OCC dispatch is built.
+// Returns the loaded DLQ meta on success so the caller does not have
+// to re-load it.
+//
+// Failure modes (all surfaced as 4xx sqsAPIError):
+//   - self-referential RedrivePolicy (defense-in-depth against records
+//     that predate the attribute-time validator),
+//   - DLQ vanished between policy-set and receive,
+//   - source queue vanished mid-redrive (DeleteQueue race),
+//   - source/DLQ queue-type mismatch (FIFO ↔ Standard) — AWS forbids
+//     this and runtime is the only place it can be enforced because
+//     the catalog accepts a RedrivePolicy that names a queue created
+//     or recreated later as a different type,
+//   - FIFO DLQ paired with a source record lacking MessageGroupId
+//     (defense in depth against malformed records that slip past the
+//     type-equality check).
+func (s *SQSServer) validateRedriveTargets(
+	ctx context.Context,
+	srcQueueName string,
+	srcRec *sqsMessageRecord,
+	policy *parsedRedrivePolicy,
+	readTS uint64,
+) (*sqsQueueMeta, error) {
+	if policy.DLQName == srcQueueName {
+		return nil, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
+			"RedrivePolicy.deadLetterTargetArn must not point at the source queue")
+	}
+	dlqMeta, dlqExists, err := s.loadQueueMetaAt(ctx, policy.DLQName, readTS)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	if !dlqExists {
+		return nil, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
+			"RedrivePolicy targets non-existent DLQ "+policy.DLQName)
+	}
+	srcMeta, srcExists, err := s.loadQueueMetaAt(ctx, srcQueueName, readTS)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	if !srcExists {
+		return nil, newSQSAPIError(http.StatusBadRequest, sqsErrQueueDoesNotExist,
+			"source queue disappeared during redrive")
+	}
+	if srcMeta.IsFIFO != dlqMeta.IsFIFO {
+		return nil, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
+			"RedrivePolicy queue-type mismatch: source and DLQ must both be FIFO or both Standard")
+	}
+	if dlqMeta.IsFIFO && srcRec.MessageGroupId == "" {
+		return nil, newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
+			"FIFO DLQ requires source records to carry MessageGroupId")
+	}
+	return dlqMeta, nil
+}
+
 // buildDLQRecord assembles the DLQ-side message record. Reset
 // ReceiveCount and FirstReceiveMillis so the DLQ consumer sees a
 // fresh delivery, not the source's bounce history.