backup: M5-3 reject non-power-of-two partition_count at load (coderabbit Major #929)

partitionForGroup masks with (PartitionCount - 1) which assumes
PartitionCount is a power of two. The live adapter validator
(adapter/sqs_partitioning.go:isPowerOfTwo) enforces this for newly-
created queues, but a malformed dump like partition_count=3 would
pass readQueueMeta as-is and then hash inconsistently: (h & 0b10)
masks bit 2 only, so partition 1 is unreachable and message routing
breaks.

Fix: in readQueueMeta, after the format_version check, fail closed
on (pub.PartitionCount > 1 && PartitionCount & (PartitionCount-1)
!= 0) with ErrSQSEncodeInvalidQueue. Caught at load time so the
operator sees the invalid dump before any record is staged.

Caller audit: readQueueMeta is the single source of meta values for
the encoder pipeline. The new gate fires before encodeQueue's
addQueueMeta, validatePartitioning, and stageMessageRecords - none
of which need to repeat the check.

Regression tests:
- TestSQSEncodeRejectsNonPowerOfTwoPartitionCount: 3/5/6/7/9/10 all
  fail closed.
- TestSQSEncodeAcceptsPowerOfTwoPartitionCount: 2/4/8/16/32 all
  pass.

Author: bootjp

internal/backup/encode_sqs.go

@@ -624,6 +624,16 @@ func (e *SQSRecordEncoder) readQueueMeta(root *os.Root, queueDir string) (sqsQue
 		return sqsQueueMetaPublic{}, errors.Wrapf(ErrSQSEncodeInvalidQueue,
 			"%s: unsupported format_version %d", rel, pub.FormatVersion)
 	}
+	// PartitionCount must be a power of two when > 1. The live
+	// validator (adapter/sqs_partitioning.go:isPowerOfTwo) enforces
+	// this so partitionFor's mask AND (h & (n-1)) is equivalent to
+	// (h % n). A malformed dump with e.g. partition_count=3 would
+	// hash inconsistently and route messages to wrong partitions.
+	// Coderabbit Major #929.
+	if pub.PartitionCount > 1 && pub.PartitionCount&(pub.PartitionCount-1) != 0 {
+		return sqsQueueMetaPublic{}, errors.Wrapf(ErrSQSEncodeInvalidQueue,
+			"%s: partition_count %d must be a power of two", rel, pub.PartitionCount)
+	}
 	return pub, nil
 }
 

internal/backup/encode_sqs_partitioned_test.go

@@ -3,6 +3,7 @@ package backup
 import (
 	"bytes"
 	"encoding/binary"
+	"fmt"
 	"testing"
 
 	"github.com/cockroachdb/errors"
@@ -373,6 +374,52 @@ func TestSQSEncodeClassicDedupKeepsLatestOnExpiredCollision(t *testing.T) {
 	}
 }
 
+// TestSQSEncodeRejectsNonPowerOfTwoPartitionCount pins coderabbit
+// Major #929. partitionForGroup masks with (PartitionCount - 1)
+// which assumes a power of two. A malformed dump like
+// partition_count=3 would mask via (h & 0b10) and route messages
+// inconsistently. readQueueMeta must fail closed.
+func TestSQSEncodeRejectsNonPowerOfTwoPartitionCount(t *testing.T) {
+	t.Parallel()
+	for _, n := range []uint32{3, 5, 6, 7, 9, 10} {
+		t.Run(fmt.Sprintf("count=%d", n), func(t *testing.T) {
+			t.Parallel()
+			in := t.TempDir()
+			writeSQSQueue(t, in, "bad.fifo",
+				[]byte(fmt.Sprintf(`{"format_version":1,"name":"bad.fifo","fifo":true,`+
+					`"visibility_timeout_seconds":30,"message_retention_seconds":345600,`+
+					`"delay_seconds":0,"partition_count":%d,"fifo_throughput_limit":"perMessageGroupId"}`, n)),
+				nil)
+			b := newSnapshotBuilder(sqsEncTS)
+			err := NewSQSRecordEncoder(in).Encode(b)
+			if !errors.Is(err, ErrSQSEncodeInvalidQueue) {
+				t.Fatalf("partition_count=%d: err = %v, want ErrSQSEncodeInvalidQueue", n, err)
+			}
+		})
+	}
+}
+
+// TestSQSEncodeAcceptsPowerOfTwoPartitionCount confirms the
+// power-of-two guard does not reject legitimate values.
+func TestSQSEncodeAcceptsPowerOfTwoPartitionCount(t *testing.T) {
+	t.Parallel()
+	for _, n := range []uint32{2, 4, 8, 16, 32} {
+		t.Run(fmt.Sprintf("count=%d", n), func(t *testing.T) {
+			t.Parallel()
+			in := t.TempDir()
+			writeSQSQueue(t, in, "ok.fifo",
+				[]byte(fmt.Sprintf(`{"format_version":1,"name":"ok.fifo","fifo":true,`+
+					`"visibility_timeout_seconds":30,"message_retention_seconds":345600,`+
+					`"delay_seconds":0,"partition_count":%d,"fifo_throughput_limit":"perQueue"}`, n)),
+				nil)
+			b := newSnapshotBuilder(sqsEncTS)
+			if err := NewSQSRecordEncoder(in).Encode(b); err != nil {
+				t.Fatalf("partition_count=%d: unexpected err = %v", n, err)
+			}
+		})
+	}
+}
+
 // TestSQSEncodeClassicDedupKeepsLatestAcrossGroups pins codex P2
 // #929 (round 2): for classic FIFO queues the live dedup key is
 // (queue, generation, dedupID) — NO group / partition. Two retained