feat(sqs): receipt-handle v2 codec for partitioned FIFO queues (Phase 3.D PR 5a)

Adds the v2 receipt-handle layout that PR 5b's send / receive
fanout will consume. Pure scaffold: the v2 encoder is added but
not yet called by any production path (PartitionCount > 1 is
still rejected by the §11 PR 2 dormancy gate, so SendMessage
never reaches a partitioned queue, so encodeReceiptHandleV2 stays
unreachable until PR 5b lifts the gate).

Splitting the codec out as a small isolated PR keeps PR 5b's
review focused on the gate-and-lift atomic sequence — codec
correctness and the gate-lift correctness are independent
concerns and a single mega-PR would conflate them.

Wire format

v1 (legacy, single partition):
  [ 0 ]    byte version = 0x01
  [ 1..9 ] uint64 queue_gen (BE)
  [ 9..25 ] 16 bytes message_id
  [ 25..41 ] 16 bytes receipt_token
  Total: 41 bytes

v2 (partitioned, PartitionCount > 1):
  [ 0 ]      byte version = 0x02
  [ 1..5 ]   uint32 partition (BE)            ← NEW
  [ 5..13 ]  uint64 queue_gen (BE)
  [ 13..29 ] 16 bytes message_id
  [ 29..45 ] 16 bytes receipt_token
  Total: 45 bytes

decodedReceiptHandle gains Version (byte) and Partition (uint32)
fields. decodeReceiptHandle dispatches by the version byte; v1
keeps Partition=0 by definition.

What does NOT change yet

- encodeReceiptHandleV2 is dormant until PR 5b's SendMessage
  partitioned-fanout dispatch wires it (gated by
  meta.PartitionCount > 1). Production traffic continues to use
  the v1 encoder verbatim — no behaviour change.
- DeleteMessage / ChangeMessageVisibility still consume
  decodedReceiptHandle without inspecting Version or Partition.
  PR 5b adds the cross-version rejection contract (a v1 handle
  against a partitioned queue → ReceiptHandleIsInvalid, and
  vice-versa).

Tests

- TestEncodeReceiptHandleV2_RoundTrip — every (partition,
  queue_gen, message_id, token) tuple round-trips with
  Version=v2 and the encoder's partition.
- TestEncodeReceiptHandleV1_StillReportsV1 — regression: v1
  encoder still produces v1-decodable handles with Partition=0.
- TestDecodeReceiptHandle_VersionDispatch — v1 and v2 produce
  distinct on-wire bytes (different version + 4-byte size delta);
  decoder picks the right layout for each.
- TestDecodeReceiptHandle_RejectsLengthMismatch — 5 cases (v1
  byte / v2 length, v2 byte / v1 length, v1 truncated, v2
  truncated, empty) all fail with the same opaque error.
- TestDecodeReceiptHandle_RejectsUnknownVersion — 0x00, 0x03,
  0x42, 0xFF all reject.
- TestEncodeReceiptHandleV2_RejectsBadInputs — short / long
  token, non-hex / short / empty id all surface as encoder errors.
- TestReceiptHandleVersionConstants_Distinct — pins the
  invariants v1 != v2, v1 == 0x01, v2 == 0x02, legacy alias
  points at v1, on-wire size constants match what encoders write.
- TestDecodeReceiptHandle_RejectsBase64Garbage — base64 errors
  surface at the right step.
- Existing TestSQSServer_ReceiptHandleCodecRoundTrip continues
  to pass unchanged (v1 path).

Self-review (per CLAUDE.md)

1. Data loss — codec only; no FSM/Pebble path. No issue.
2. Concurrency — no shared state; pure encode/decode functions.
   No issue.
3. Performance — encode/decode are O(handle size) ≤ 45 bytes.
   The version-byte dispatch is one switch. No issue.
4. Data consistency — v1 and v2 are length-distinct, version-byte-
   distinct, and decoder rejects wrong-length blobs. A v1 client
   parsing a v2 handle (or vice versa) fails closed at the
   length check — pinned by the 5-case rejection test. No issue.
5. Test coverage — 8 new tests across the codec contract surface;
   existing v1 round-trip test continues to pass.

Author: bootjp

adapter/sqs_messages.go

@@ -43,9 +43,27 @@ const (
 	// has <1% tail-latency overhead, large enough that an empty queue
 	// does not spin.
 	sqsLongPollInterval = 200 * time.Millisecond
-	// Version byte prefixed to encoded receipt handles. Bumped when the
-	// on-wire handle format changes so old handles fail to decode loudly.
-	sqsReceiptHandleVersion = byte(0x01)
+	// sqsReceiptHandleVersion1 is the version byte for the legacy
+	// (single-partition) receipt-handle layout — pre-PR-5 handles
+	// and post-PR-5 handles for queues with PartitionCount <= 1.
+	// Bumped when the on-wire handle format changes so old handles
+	// fail to decode loudly.
+	sqsReceiptHandleVersion1 = byte(0x01)
+	// sqsReceiptHandleVersion2 is the version byte for partitioned
+	// FIFO handles (PartitionCount > 1). v2 carries the partition
+	// index so DeleteMessage / ChangeMessageVisibility can dispatch
+	// to the right partition's keyspace without re-running the
+	// partitionFor hash. The version byte discriminates: a v1 handle
+	// against a partitioned queue is rejected, a v2 handle against
+	// a non-partitioned queue is rejected — both with
+	// ReceiptHandleIsInvalid (codex P1 round-2 on PR #715 documents
+	// the cross-version rejection contract).
+	sqsReceiptHandleVersion2 = byte(0x02)
+	// sqsReceiptHandleVersion is the legacy const name preserved as
+	// an alias of v1 so existing call sites in this file keep
+	// compiling. New call sites SHOULD use sqsReceiptHandleVersion1
+	// (or sqsReceiptHandleVersion2) explicitly.
+	sqsReceiptHandleVersion = sqsReceiptHandleVersion1
 	// Byte sizes used when pre-sizing key buffers. The exact value is not
 	// critical; it only avoids one append growth for typical queue/ID
 	// lengths.
@@ -213,16 +231,27 @@ func newReceiptToken() ([]byte, error) {
 	return buf, nil
 }
 
-// encodeReceiptHandle packs (queue_gen, message_id, receipt_token) into a
-// single opaque blob. Format:
+// sqsReceiptHandleV1Size is the on-wire byte length of a v1
+// receipt handle: 1 byte version + 8 byte queue_gen + 16 byte
+// message_id + 16 byte receipt_token = 41 bytes.
+const sqsReceiptHandleV1Size = 1 + 8 + sqsMessageIDBytes + sqsReceiptTokenBytes
+
+// sqsReceiptHandleV2Size is the on-wire byte length of a v2
+// (partitioned) receipt handle: v1 + 4 bytes for the partition
+// uint32 inserted between version and queue_gen = 45 bytes.
+const sqsReceiptHandleV2Size = 1 + 4 + 8 + sqsMessageIDBytes + sqsReceiptTokenBytes
+
+// encodeReceiptHandle packs (queue_gen, message_id, receipt_token) into
+// a single opaque v1 blob. Used by SendMessage on a NON-partitioned
+// FIFO queue and on Standard queues. Format:
 //
-//	[ 0 ] byte version = 0x01
+//	[ 0 ]    byte version = 0x01
 //	[ 1..9 ] uint64 queue_gen (BE)
 //	[ 9..25 ] 16 bytes message_id (raw bytes from hex decode)
 //	[ 25..41 ] 16 bytes receipt_token
 //
-// The result is base64-urlsafe (no padding) so it passes through JSON and
-// HTTP query parameters untouched.
+// The result is base64-urlsafe (no padding) so it passes through
+// JSON and HTTP query parameters untouched.
 func encodeReceiptHandle(queueGen uint64, messageIDHex string, receiptToken []byte) (string, error) {
 	if len(receiptToken) != sqsReceiptTokenBytes {
 		return "", errors.New("receipt token has wrong length")
@@ -231,35 +260,113 @@ func encodeReceiptHandle(queueGen uint64, messageIDHex string, receiptToken []by
 	if err != nil || len(idBytes) != sqsMessageIDBytes {
 		return "", errors.New("message id has wrong format")
 	}
-	buf := make([]byte, 0, 1+8+sqsMessageIDBytes+sqsReceiptTokenBytes)
-	buf = append(buf, sqsReceiptHandleVersion)
+	buf := make([]byte, 0, sqsReceiptHandleV1Size)
+	buf = append(buf, sqsReceiptHandleVersion1)
 	buf = appendU64(buf, queueGen)
 	buf = append(buf, idBytes...)
 	buf = append(buf, receiptToken...)
 	return base64.RawURLEncoding.EncodeToString(buf), nil
 }
 
+// encodeReceiptHandleV2 packs (partition, queue_gen, message_id,
+// receipt_token) for partitioned FIFO queues. Used by SendMessage
+// on a partitioned queue (PartitionCount > 1). Format:
+//
+//	[ 0 ]      byte version = 0x02
+//	[ 1..5 ]   uint32 partition (BE)
+//	[ 5..13 ]  uint64 queue_gen (BE)
+//	[ 13..29 ] 16 bytes message_id
+//	[ 29..45 ] 16 bytes receipt_token
+//
+// DeleteMessage / ChangeMessageVisibility use the partition field
+// to route the operation to the right partition's keyspace
+// without re-running partitionFor (the original MessageGroupId is
+// not in the handle).
+func encodeReceiptHandleV2(partition uint32, queueGen uint64, messageIDHex string, receiptToken []byte) (string, error) {
+	if len(receiptToken) != sqsReceiptTokenBytes {
+		return "", errors.New("receipt token has wrong length")
+	}
+	idBytes, err := hex.DecodeString(messageIDHex)
+	if err != nil || len(idBytes) != sqsMessageIDBytes {
+		return "", errors.New("message id has wrong format")
+	}
+	buf := make([]byte, 0, sqsReceiptHandleV2Size)
+	buf = append(buf, sqsReceiptHandleVersion2)
+	buf = appendU32(buf, partition)
+	buf = appendU64(buf, queueGen)
+	buf = append(buf, idBytes...)
+	buf = append(buf, receiptToken...)
+	return base64.RawURLEncoding.EncodeToString(buf), nil
+}
+
+// decodedReceiptHandle is the parsed shape of a receipt handle.
+// Version reports which on-wire format the handle was in:
+//   - 0x01 (sqsReceiptHandleVersion1): legacy single-partition
+//     handle. Partition is 0 by definition.
+//   - 0x02 (sqsReceiptHandleVersion2): partitioned-FIFO handle.
+//     Partition is the partition index the message was sent to.
+//
+// Callers that route by partition (DeleteMessage,
+// ChangeMessageVisibility on a partitioned queue) MUST inspect
+// Version before consulting Partition — a v1 handle's zero
+// Partition is the LEGACY meaning ("single partition"), not
+// "partition 0 of N". A queue's PartitionCount at the moment the
+// caller acts must agree with the handle's version, or the
+// dispatch is rejected with ReceiptHandleIsInvalid.
 type decodedReceiptHandle struct {
+	Version         byte
+	Partition       uint32
 	QueueGeneration uint64
 	MessageIDHex    string
 	ReceiptToken    []byte
 }
 
+// decodeReceiptHandle inspects the version byte to dispatch to
+// the v1 or v2 layout. Length and version mismatches both surface
+// as the same opaque error so a malicious caller cannot use the
+// error message to probe the format.
 func decodeReceiptHandle(raw string) (*decodedReceiptHandle, error) {
 	b, err := base64.RawURLEncoding.DecodeString(raw)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
-	want := 1 + 8 + sqsMessageIDBytes + sqsReceiptTokenBytes
-	if len(b) != want || b[0] != sqsReceiptHandleVersion {
+	if len(b) == 0 {
+		return nil, errors.New("receipt handle length or version mismatch")
+	}
+	switch b[0] {
+	case sqsReceiptHandleVersion1:
+		return decodeReceiptHandleV1(b)
+	case sqsReceiptHandleVersion2:
+		return decodeReceiptHandleV2(b)
+	default:
 		return nil, errors.New("receipt handle length or version mismatch")
 	}
-	out := &decodedReceiptHandle{
+}
+
+func decodeReceiptHandleV1(b []byte) (*decodedReceiptHandle, error) {
+	if len(b) != sqsReceiptHandleV1Size {
+		return nil, errors.New("receipt handle length or version mismatch")
+	}
+	return &decodedReceiptHandle{
+		Version:         sqsReceiptHandleVersion1,
+		Partition:       0,
 		QueueGeneration: binary.BigEndian.Uint64(b[1:9]),
 		MessageIDHex:    hex.EncodeToString(b[9 : 9+sqsMessageIDBytes]),
 		ReceiptToken:    bytes.Clone(b[9+sqsMessageIDBytes:]),
+	}, nil
+}
+
+func decodeReceiptHandleV2(b []byte) (*decodedReceiptHandle, error) {
+	if len(b) != sqsReceiptHandleV2Size {
+		return nil, errors.New("receipt handle length or version mismatch")
 	}
-	return out, nil
+	return &decodedReceiptHandle{
+		Version:         sqsReceiptHandleVersion2,
+		Partition:       binary.BigEndian.Uint32(b[1:5]),
+		QueueGeneration: binary.BigEndian.Uint64(b[5:13]),
+		MessageIDHex:    hex.EncodeToString(b[13 : 13+sqsMessageIDBytes]),
+		ReceiptToken:    bytes.Clone(b[13+sqsMessageIDBytes:]),
+	}, nil
 }
 
 // ------------------------ input decoding ------------------------