fix(sqs): assign FIFO seq on FIFO->FIFO redrive + reject mixed attribute values

Two findings from the latest Codex review on `cdb3c87` that the
prior "Ready to merge" verdicts had not addressed.

P1 — FIFO->FIFO redrive committed `SequenceNumber=0` and never
advanced the DLQ's sqsQueueSeqKey (`sqs_redrive.go:265-272, 213`).
Once `cdb3c87` enabled FIFO->FIFO redrives by enforcing type
equality, the SequenceNumber path became reachable and showed:
- the DLQ record carried the zero-value `SequenceNumber` (AWS
  surfaces this verbatim; AWS sequences start at 1, never 0),
- the DLQ's per-queue sequence counter was never incremented, so a
  subsequent normal FIFO send to the DLQ assigned a number lower
  than the redriven message — non-monotonic to consumers, breaking
  the FIFO contract,
- two concurrent FIFO->FIFO redrives both wrote `SequenceNumber=0`
  with no OCC conflict because `sqsQueueSeqKey(policy.DLQName)` was
  not on `ReadKeys`.

Fix: when `dlqMeta.IsFIFO`, `redriveCandidateToDLQ` now loads the
DLQ's seq counter at `readTS`, computes `nextSeq = prev+1`, threads
the value through `buildDLQRecord` (where it is stamped onto the
record before encoding) and `buildRedriveOps` (where the seq Put is
added to `Elems` and `sqsQueueSeqKey(policy.DLQName)` is added to
`ReadKeys`). Standard DLQs are unchanged: `dlqSeq=0` flows through
unused.

P2 — `validateOneMessageAttribute` accepted Binary attributes that
also carried a non-empty `StringValue` (`sqs_messages.go:1523-1527`).
AWS requires each MessageAttributeValue to populate exactly one of
{StringValue, BinaryValue}; submitting both is an
`InvalidParameterValue`. The previous validator only checked the
type-required field was non-empty and silently let the foreign
field through, where it would be persisted into the record and
round-tripped on ReceiveMessage with mismatched MD5 hashes.

Fix: extracted the type-specific value-pair check into
`validateMessageAttributeValuePair` and added the symmetric guards
— Binary rejects `StringValue != ""`, String/Number rejects
`len(BinaryValue) > 0`. The split also keeps
`validateOneMessageAttribute` under the cyclop budget.

Tests:

- `TestSQSServer_FifoFifoRedriveAssignsSequenceNumber`
  (`sqs_extra_test.go`): create FIFO source + FIFO DLQ, send poison,
  redrive on the second receive, assert the DLQ message's
  SequenceNumber > 0, then send a fresh FIFO message to the DLQ and
  assert its SequenceNumber == redrivenSeq + 1 (monotonic
  continuation, the invariant the fix guarantees).
- `TestSQSServer_SendMessageRejectsBinaryWithStringValue`
  (`sqs_extra_test.go`): table-driven over the three illegal
  combinations (Binary + non-empty StringValue, String + non-empty
  BinaryValue, Number + non-empty BinaryValue); each must return
  HTTP 400 / InvalidAttributeValue.

Per CLAUDE.md self-review:
1. Data loss — None. The poison message body / attributes flow
   unchanged; only the SequenceNumber field and DLQ counter change.
2. Concurrency — `sqsQueueSeqKey(policy.DLQName)` is now on
   ReadKeys, so two concurrent FIFO->FIFO redrives serialise on
   write-write-conflict, same as concurrent FIFO sends.
3. Performance — One extra `loadFifoSequence` per FIFO redrive;
   negligible vs. the existing `validateRedriveTargets` reads.
4. Data consistency — DLQ SequenceNumber now monotonic across both
   redrive and normal-send paths; no two messages can share a
   sequence number on the same FIFO queue.
5. Test coverage — Both fixes ship with co-located regression
   tests; the existing FIFO-related tests
   (`TestSQSServer_FifoSequenceNumberMonotonic`,
   `TestSQSServer_DLQRedriveOnMaxReceiveCount`,
   `TestSQSServer_RedrivePolicy*`) remain green.

Author: bootjp

adapter/sqs_extra_test.go

@@ -1746,3 +1746,164 @@ func TestSQSServer_TagQueueRequiresTags(t *testing.T) {
 		t.Fatalf("untag without TagKeys: %d %v", status, out)
 	}
 }
+
+// TestSQSServer_FifoFifoRedriveAssignsSequenceNumber pins the round-N
+// Codex P1 fix on `cdb3c87` redrive: a FIFO source whose RedrivePolicy
+// targets a FIFO DLQ used to commit the DLQ record with
+// SequenceNumber = 0 (zero-value, not even AWS's "starts at 1"
+// invariant), and the DLQ's per-queue sequence counter
+// (sqsQueueSeqKey) was never advanced. Consumers reading the DLQ saw
+// 0 verbatim, and a normal FIFO send to the DLQ later produced a
+// number lower than the redriven message — non-monotonic per AWS's
+// FIFO contract. The fix loads the DLQ seq at readTS, increments it,
+// stamps it onto the DLQ record, and includes the seq Put in the OCC
+// transaction. This test asserts both halves: (a) the redriven DLQ
+// record carries SequenceNumber = 1, (b) a subsequent FIFO send to
+// the DLQ carries SequenceNumber = 2.
+func TestSQSServer_FifoFifoRedriveAssignsSequenceNumber(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	// FIFO DLQ.
+	_, out := callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName":  "dlq-fifo.fifo",
+		"Attributes": map[string]string{"FifoQueue": "true"},
+	})
+	dlqURL, _ := out["QueueUrl"].(string)
+	if dlqURL == "" {
+		t.Fatalf("FIFO DLQ create failed: %v", out)
+	}
+
+	// FIFO source pointing at the FIFO DLQ. Both queues are FIFO so
+	// the type-equality guard added in cdb3c87 lets the redrive
+	// proceed; this test exercises the SequenceNumber assignment that
+	// became reachable only with that change.
+	policy := `{"deadLetterTargetArn":"arn:aws:sqs:us-east-1:000000000000:dlq-fifo.fifo","maxReceiveCount":1}`
+	_, out = callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName": "src-fifo.fifo",
+		"Attributes": map[string]string{
+			"FifoQueue":     "true",
+			"RedrivePolicy": policy,
+		},
+	})
+	srcURL, _ := out["QueueUrl"].(string)
+
+	// Send a poison message to the FIFO source.
+	_, _ = callSQS(t, node, sqsSendMessageTarget, map[string]any{
+		"QueueUrl":               srcURL,
+		"MessageBody":            "poison",
+		"MessageGroupId":         "g1",
+		"MessageDeduplicationId": "d-poison",
+	})
+
+	// First receive bumps ReceiveCount to 1 == maxReceiveCount;
+	// second receive (after visibility expiry) triggers the redrive.
+	_, _ = 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,
+	})
+
+	// DLQ now has the moved message. SequenceNumber must be > 0;
+	// pre-fix it was 0 (zero-value, never set).
+	_, out = callSQS(t, node, sqsReceiveMessageTarget, map[string]any{
+		"QueueUrl":            dlqURL,
+		"MaxNumberOfMessages": 1,
+		"VisibilityTimeout":   60,
+	})
+	msgs, _ := out["Messages"].([]any)
+	if len(msgs) != 1 {
+		t.Fatalf("DLQ expected 1 redriven message, got %d (%v)", len(msgs), out)
+	}
+	moved, _ := msgs[0].(map[string]any)
+	movedAttrs, _ := moved["Attributes"].(map[string]any)
+	movedSeqStr, _ := movedAttrs["SequenceNumber"].(string)
+	movedSeq, _ := strconv.ParseUint(movedSeqStr, 10, 64)
+	if movedSeq == 0 {
+		t.Fatalf("redriven DLQ message has SequenceNumber=0 (regression: counter not advanced); attrs=%v", movedAttrs)
+	}
+
+	// Second half: a normal FIFO send to the DLQ must observe the
+	// advanced counter and assign movedSeq + 1, not start over from
+	// 1 (which would put two messages with the same sequence on the
+	// queue, the exact bug the fix is preventing).
+	follow := sendFifoMessage(t, node, dlqURL, "g-dlq", "d-follow", "follow")
+	if follow != movedSeq+1 {
+		t.Fatalf("DLQ FIFO send after redrive: SequenceNumber=%d, want %d (= %d + 1)", follow, movedSeq+1, movedSeq)
+	}
+}
+
+// TestSQSServer_SendMessageRejectsBinaryWithStringValue pins the
+// round-N Codex P2 fix: AWS requires a MessageAttributeValue to
+// populate exactly one of {StringValue, BinaryValue}. The previous
+// validator only checked that BinaryValue was non-empty for Binary
+// types; an attribute carrying both fields would be persisted into
+// the record (and round-tripped on ReceiveMessage), which is not
+// AWS behavior and would surface mismatched MD5 hashes downstream.
+// The symmetric String/Number + non-empty BinaryValue case is also
+// asserted.
+func TestSQSServer_SendMessageRejectsBinaryWithStringValue(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+	queueURL := createSQSQueueForTest(t, node, "binary-string-mix")
+
+	cases := []struct {
+		name  string
+		attrs map[string]any
+	}{
+		{
+			name: "Binary with non-empty StringValue",
+			attrs: map[string]any{
+				"X": map[string]any{
+					"DataType":    "Binary",
+					"BinaryValue": []byte{0x01, 0x02, 0x03},
+					"StringValue": "stowaway",
+				},
+			},
+		},
+		{
+			name: "String with non-empty BinaryValue",
+			attrs: map[string]any{
+				"X": map[string]any{
+					"DataType":    "String",
+					"StringValue": "ok",
+					"BinaryValue": []byte{0x01},
+				},
+			},
+		},
+		{
+			name: "Number with non-empty BinaryValue",
+			attrs: map[string]any{
+				"X": map[string]any{
+					"DataType":    "Number",
+					"StringValue": "1",
+					"BinaryValue": []byte{0x01},
+				},
+			},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			status, out := callSQS(t, node, sqsSendMessageTarget, map[string]any{
+				"QueueUrl":          queueURL,
+				"MessageBody":       "body",
+				"MessageAttributes": tc.attrs,
+			})
+			if status != http.StatusBadRequest {
+				t.Fatalf("expected 400, got %d (%v)", status, out)
+			}
+			if got, _ := out["__type"].(string); got != sqsErrInvalidAttributeValue {
+				t.Fatalf("error type: %q want %q (%v)", got, sqsErrInvalidAttributeValue, out)
+			}
+		})
+	}
+}

adapter/sqs_messages.go

@@ -1514,17 +1514,37 @@ func validateOneMessageAttribute(name string, v sqsMessageAttributeValue) error
 	if i := strings.Index(v.DataType, "."); i >= 0 {
 		base = v.DataType[:i]
 	}
+	return validateMessageAttributeValuePair(name, base, v)
+}
+
+// validateMessageAttributeValuePair enforces "exactly one value field
+// populated, matching the DataType" on a MessageAttributeValue. AWS
+// rejects an attribute that carries both StringValue and BinaryValue
+// (or that carries the wrong one for its DataType); without these
+// guards a malformed client could persist bytes into the record that
+// then round-trip on ReceiveMessage, producing mismatched MD5 hashes
+// downstream. Pulled out of validateOneMessageAttribute so that
+// function stays under the cyclop budget.
+func validateMessageAttributeValuePair(name, base string, v sqsMessageAttributeValue) error {
 	switch base {
 	case "String", "Number":
 		if v.StringValue == "" {
 			return newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
 				"MessageAttribute "+name+" requires StringValue")
 		}
+		if len(v.BinaryValue) > 0 {
+			return newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
+				"MessageAttribute "+name+" must not include BinaryValue for "+base+" type")
+		}
 	case sqsAttributeBaseTypeBinary:
 		if len(v.BinaryValue) == 0 {
 			return newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
 				"MessageAttribute "+name+" requires BinaryValue")
 		}
+		if v.StringValue != "" {
+			return newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
+				"MessageAttribute "+name+" must not include StringValue for Binary type")
+		}
 	default:
 		return newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
 			"MessageAttribute "+name+" has unsupported DataType "+v.DataType)

adapter/sqs_redrive.go

@@ -3,6 +3,7 @@ package adapter
 import (
 	"context"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -125,11 +126,28 @@ func (s *SQSServer) redriveCandidateToDLQ(
 	if err != nil {
 		return false, err
 	}
-	dlqRec, dlqRecordBytes, err := buildDLQRecord(srcRec, dlqMeta, srcArn)
+	// FIFO DLQs require the redrive write to participate in the
+	// per-queue SequenceNumber sequence, otherwise the DLQ record
+	// is committed with SequenceNumber=0 (AWS surfaces this
+	// verbatim, and 0 violates AWS's invariant that sequences
+	// start at 1) and the next normal FIFO send to the DLQ assigns
+	// a number lower than the redriven message — non-monotonic to
+	// consumers. Load the seq snapshot at readTS, increment, and
+	// pass it into both buildDLQRecord (encoded onto the record)
+	// and buildRedriveOps (Put + ReadKeys fence).
+	var dlqSeq uint64
+	if dlqMeta.IsFIFO {
+		prevSeq, err := s.loadFifoSequence(ctx, policy.DLQName, readTS)
+		if err != nil {
+			return false, err
+		}
+		dlqSeq = prevSeq + 1
+	}
+	dlqRec, dlqRecordBytes, err := buildDLQRecord(srcRec, dlqMeta, srcArn, dlqSeq)
 	if err != nil {
 		return false, err
 	}
-	req, err := s.buildRedriveOps(ctx, srcQueueName, srcGen, cand, srcDataKey, srcRec, policy, dlqMeta, dlqRec, dlqRecordBytes, readTS)
+	req, err := s.buildRedriveOps(ctx, srcQueueName, srcGen, cand, srcDataKey, srcRec, policy, dlqMeta, dlqRec, dlqRecordBytes, dlqSeq, readTS)
 	if err != nil {
 		return false, err
 	}
@@ -200,7 +218,16 @@ func (s *SQSServer) validateRedriveTargets(
 // 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.
-func buildDLQRecord(srcRec *sqsMessageRecord, dlqMeta *sqsQueueMeta, srcArn string) (*sqsMessageRecord, []byte, error) {
+//
+// dlqSeq is the SequenceNumber to assign on the DLQ record, computed
+// by the caller as `loadFifoSequence(dlq) + 1` for FIFO DLQs and
+// passed as 0 for Standard DLQs (the field is unused in that case).
+// The seq must be the same value the caller will Put into the DLQ's
+// sqsQueueSeqKey inside the same OCC transaction (see buildRedriveOps);
+// otherwise the redriven message and the on-disk counter disagree
+// and a later FIFO send to the DLQ produces a non-monotonic
+// SequenceNumber.
+func buildDLQRecord(srcRec *sqsMessageRecord, dlqMeta *sqsQueueMeta, srcArn string, dlqSeq uint64) (*sqsMessageRecord, []byte, error) {
 	dlqMsgID, err := newMessageIDHex()
 	if err != nil {
 		return nil, nil, errors.WithStack(err)
@@ -227,6 +254,7 @@ func buildDLQRecord(srcRec *sqsMessageRecord, dlqMeta *sqsQueueMeta, srcArn stri
 		MessageGroupId:         srcRec.MessageGroupId,
 		MessageDeduplicationId: srcRec.MessageDeduplicationId,
 		DeadLetterSourceArn:    srcArn,
+		SequenceNumber:         dlqSeq,
 	}
 	body, err := encodeSQSMessageRecord(rec)
 	if err != nil {
@@ -239,6 +267,14 @@ func buildDLQRecord(srcRec *sqsMessageRecord, dlqMeta *sqsQueueMeta, srcArn stri
 // atomically removes the source's keyspace and writes the DLQ
 // version. The FIFO group-lock release branch lives here so the
 // caller stays under the cyclomatic budget.
+//
+// dlqSeq is non-zero only when the DLQ is FIFO (per the caller's
+// pre-load via loadFifoSequence). When non-zero, the txn additionally
+// reads sqsQueueSeqKey(policy.DLQName) — guarding against a
+// concurrent FIFO send / redrive racing for the same sequence — and
+// writes the new value back. dlqRec.SequenceNumber is already set to
+// dlqSeq inside buildDLQRecord; this function is responsible only for
+// the OCC plumbing.
 func (s *SQSServer) buildRedriveOps(
 	ctx context.Context,
 	srcQueueName string,
@@ -250,6 +286,7 @@ func (s *SQSServer) buildRedriveOps(
 	dlqMeta *sqsQueueMeta,
 	dlqRec *sqsMessageRecord,
 	dlqRecordBytes []byte,
+	dlqSeq uint64,
 	readTS uint64,
 ) (*kv.OperationGroup[kv.OP], error) {
 	now := dlqRec.SendTimestampMillis
@@ -270,6 +307,15 @@ func (s *SQSServer) buildRedriveOps(
 		{Op: kv.Put, Key: dlqVisKey, Value: []byte(dlqRec.MessageID)},
 		{Op: kv.Put, Key: dlqByAgeKey, Value: []byte(dlqRec.MessageID)},
 	}
+	if dlqMeta.IsFIFO {
+		seqKey := sqsQueueSeqKey(policy.DLQName)
+		readKeys = append(readKeys, seqKey)
+		elems = append(elems, &kv.Elem[kv.OP]{
+			Op:    kv.Put,
+			Key:   seqKey,
+			Value: []byte(strconv.FormatUint(dlqSeq, 10)),
+		})
+	}
 	if srcRec.MessageGroupId != "" {
 		lockKey := sqsMsgGroupKey(srcQueueName, srcGen, srcRec.MessageGroupId)
 		lock, err := s.loadFifoGroupLock(ctx, srcQueueName, srcGen, srcRec.MessageGroupId, readTS)