sqs(catalog,capability): idempotent gate placement + sanitized public message (PR #734, round 1)

Two PR #734 review findings addressed in one commit because they
touch the same code path:

1. Codex P1 — Check existing queue before enforcing HTFIFO peer gate

   The previous placement of validateHTFIFOCapability inside
   createQueueCore (before createQueueWithRetry) ran the gate on
   EVERY CreateQueue call with PartitionCount > 1, including
   idempotent retries on an already-existing queue with identical
   attributes. A transient peer poll failure (timeout / unreachable
   / malformed health) during such a retry would then return
   InvalidAttributeValue instead of the AWS-correct 200 OK, breaking
   create-or-get clients during partial outages or rolling upgrades.

   Fix: move the gate INTO tryCreateQueueOnce after the existence
   check and BEFORE the OCC dispatch. The order in tryCreateQueueOnce
   is now:

     1. loadQueueMetaAt — check if queue exists at the snapshot
     2. exists + matching attrs → return (true, nil) idempotent OK
     3. exists + different attrs → return QueueNameExists
     4. validateHTFIFOCapability — runs ONLY on the genuine create
        path
     5. loadQueueGenerationAt + dispatch the create

   The gate may run more than once across OCC retries (each retry
   that gets to "queue still missing" re-polls), but every retry
   that hits an existing-queue match short-circuits before the gate
   runs — so idempotent CreateQueue under a partial cluster outage
   stays AWS-correct.

   Caller audit: validateHTFIFOCapability has exactly one production
   caller (now tryCreateQueueOnce); both the JSON handler
   (createQueue → createQueueCore → createQueueWithRetry →
   tryCreateQueueOnce) and the query-protocol handler
   (sqs_query_protocol.go: 182 → createQueueCore → …) reach it
   through that one path. Move is symmetric — no caller observes a
   semantic change for a queue that genuinely needs to be created;
   only the existing-queue path stops paying the gate cost.

2. CodeRabbit major — Don't send raw peer poll details back to caller

   buildHTFIFOCapabilityRejection's output (peer addresses + raw
   poller error text) was returned to the wire layer as the
   InvalidAttributeValue message, leaking cluster topology to any
   authenticated CreateQueue caller. This conflicts with the
   stricter error-redaction policy already used elsewhere in
   sqs_catalog.go.

   Fix: the wire-level rejection is now the sanitized constant
   htfifoCapabilityRejectionPublic ("PartitionCount > 1 requires
   every cluster peer to advertise the htfifo capability via
   /sqs_health; one or more peers did not — see server logs for
   details"). The full per-peer detail goes to slog.Warn with
   structured fields (queueName, partitionCount, peerCount, detail)
   so an operator triaging the rolling upgrade can read the failing
   peer addresses + reasons from the server logs without rerunning
   the poll out-of-band.

   Renamed buildHTFIFOCapabilityRejection →
   formatHTFIFOCapabilityReportForLog to make its server-side-only
   contract obvious at the call site.

Test changes:

  - New TestSQSServer_HTFIFO_CapabilityGate_IsIdempotentOnExistingQueue
    (wire-level): creates a partitioned queue on a single-node
    cluster (gate vacuously passes), poisons leaderSQS with an
    unreachable address, then re-creates the same queue with
    identical attrs and expects 200; finally creates a NEW queue
    with the poisoned peer map and expects the 400 (proves the
    gate is still in effect for genuine creates).
  - New TestValidateHTFIFOCapability_PublicMessageDoesNotLeakPeerDetails:
    pins the sanitization contract — the wire-level message must
    equal htfifoCapabilityRejectionPublic exactly, never contain
    a peer host:port.
  - Updated TestValidateHTFIFOCapability_RejectsWhenOnePeerLacksCapability
    and TestValidateHTFIFOCapability_RejectsWhenPeerUnreachable
    to assert the sanitized constant + NotContains on the peer
    address.
  - Renamed TestBuildHTFIFOCapabilityRejection_ShapesOperatorMessage
    → TestFormatHTFIFOCapabilityReportForLog_ShapesServerSideDetail
    to match the renamed helper; assertion that the helper output
    is server-side-only (no client wire surface assertion here).

Below threshold (intentionally not addressed in this round):
  - Gemini medium on collectSQSPeers concurrency: leaderSQS is
    only mutated at SQSServer construction (WithSQSLeaderMap), not
    at request time. Gemini's own comment acknowledges this.
  - Gemini medium on caching the capability status: CreateQueue is
    a rare control-plane operation; caching adds a stale-window
    failure mode (a cluster that already lost a peer would still
    accept a partitioned queue while the cache is warm). Pure
    performance suggestion, not correctness.

Author: bootjp

adapter/sqs_capability_gate.go

@@ -2,11 +2,21 @@ package adapter
 
 import (
 	"context"
+	"log/slog"
 	"net/http"
 	"sort"
 	"strings"
 )
 
+// htfifoCapabilityRejectionPublic is the sanitized client-facing
+// reason returned from validateHTFIFOCapability when the cluster
+// poll fails. Per CodeRabbit major review: peer addresses and raw
+// poller error text MUST NOT leak to authenticated clients (the
+// CreateQueue surface is part of the public AWS-shaped API), so
+// the wire-level message is intentionally generic and the per-peer
+// detail goes to slog.Warn for operator triage.
+const htfifoCapabilityRejectionPublic = "PartitionCount > 1 requires every cluster peer to advertise the htfifo capability via /sqs_health; one or more peers did not — see server logs for details"
+
 // validateHTFIFOCapability is the §11 PR 5b-3 gate that replaced the
 // PR 2 dormancy reject. CreateQueue calls this on every request; it
 // is a no-op for legacy / single-partition meta and the full
@@ -53,8 +63,18 @@ func (s *SQSServer) validateHTFIFOCapability(ctx context.Context, requested *sqs
 	}
 	report := PollSQSHTFIFOCapability(ctx, peers, PollerConfig{})
 	if report == nil || !report.AllAdvertise {
+		// Log the full per-peer detail for operator triage. The
+		// client-visible message stays generic (no peer addresses,
+		// no raw poller error text) so the CreateQueue surface
+		// does not leak cluster topology to authenticated callers
+		// — CodeRabbit major review on PR #734.
+		slog.Warn("sqs: htfifo capability gate rejected partitioned CreateQueue",
+			"queueName", requested.Name,
+			"partitionCount", requested.PartitionCount,
+			"peerCount", len(peers),
+			"detail", formatHTFIFOCapabilityReportForLog(report))
 		return newSQSAPIError(http.StatusBadRequest, sqsErrInvalidAttributeValue,
-			buildHTFIFOCapabilityRejection(report))
+			htfifoCapabilityRejectionPublic)
 	}
 	return nil
 }
@@ -86,15 +106,17 @@ func (s *SQSServer) collectSQSPeers() []string {
 	return peers
 }
 
-// buildHTFIFOCapabilityRejection composes the operator-facing message
-// for a failed capability poll. Lists the peers that did not
-// advertise htfifo (with the per-peer Error or "missing capability"
-// reason) so the operator can fix the rolling-upgrade lag without
-// rerunning the poll out-of-band. Order matches report.Peers, which
-// matches collectSQSPeers' sorted input order — deterministic.
-func buildHTFIFOCapabilityRejection(report *HTFIFOCapabilityReport) string {
+// formatHTFIFOCapabilityReportForLog composes the per-peer detail
+// surfaced to slog.Warn when the gate rejects. Lists each failing
+// peer's address and reason (per-peer Error or "missing capability")
+// so an operator triaging a partial-rolling-upgrade cluster can fix
+// the lag from the server logs without rerunning the poll
+// out-of-band. NEVER returned to the client — that path uses the
+// sanitized htfifoCapabilityRejectionPublic constant. Order matches
+// report.Peers, which matches collectSQSPeers' sorted input order —
+// deterministic so log lines diff cleanly across runs.
+func formatHTFIFOCapabilityReportForLog(report *HTFIFOCapabilityReport) string {
 	var b strings.Builder
-	b.WriteString("PartitionCount > 1 requires every cluster peer to advertise the htfifo capability via /sqs_health; the following peers did not: ")
 	if report == nil {
 		b.WriteString("(no report)")
 		return b.String()
@@ -120,7 +142,7 @@ func buildHTFIFOCapabilityRejection(report *HTFIFOCapabilityReport) string {
 	if first {
 		// Defensive: AllAdvertise was false but no peer surfaced a
 		// reason. Should never happen, but emit a non-empty hint
-		// rather than a truncated message ending in a colon.
+		// rather than a truncated empty string.
 		b.WriteString("(unknown peer)")
 	}
 	return b.String()

adapter/sqs_capability_gate_test.go

@@ -112,10 +112,10 @@ func TestValidateHTFIFOCapability_RejectsWhenOnePeerLacksCapability(t *testing.T
 	require.True(t, ok, "must surface as sqsAPIError so the wire layer maps to InvalidAttributeValue, got %T", err)
 	require.Equal(t, http.StatusBadRequest, apiErr.status)
 	require.Equal(t, sqsErrInvalidAttributeValue, apiErr.errorType)
-	require.Contains(t, apiErr.message, "every cluster peer to advertise the htfifo capability",
-		"message must explain the gate so the operator knows what to fix")
-	require.Contains(t, apiErr.message, oldAddr,
-		"the offending peer must appear in the message, got %q", apiErr.message)
+	require.Equal(t, htfifoCapabilityRejectionPublic, apiErr.message,
+		"client message must be the sanitized constant — peer addresses live in server logs (CodeRabbit major)")
+	require.NotContains(t, apiErr.message, oldAddr,
+		"peer host:port MUST NOT leak through the wire-level rejection")
 }
 
 // TestValidateHTFIFOCapability_RejectsWhenPeerUnreachable pins
@@ -145,8 +145,10 @@ func TestValidateHTFIFOCapability_RejectsWhenPeerUnreachable(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, http.StatusBadRequest, apiErr.status)
 	require.Equal(t, sqsErrInvalidAttributeValue, apiErr.errorType)
-	require.Contains(t, apiErr.message, deadAddr,
-		"unreachable peer must be named in the rejection message")
+	require.Equal(t, htfifoCapabilityRejectionPublic, apiErr.message,
+		"client message must be the sanitized constant — transport error text lives in server logs (CodeRabbit major)")
+	require.NotContains(t, apiErr.message, deadAddr,
+		"peer host:port MUST NOT leak through the wire-level rejection")
 }
 
 // TestCollectSQSPeers_Deterministic pins the helper's order +
@@ -174,11 +176,15 @@ func TestCollectSQSPeers_Deterministic(t *testing.T) {
 	require.Empty(t, (&SQSServer{}).collectSQSPeers())
 }
 
-// TestBuildHTFIFOCapabilityRejection_ShapesOperatorMessage pins the
-// rejection-message shape so a future refactor cannot accidentally
-// truncate the per-peer detail. Each failing peer must contribute
-// a "(reason)" suffix; peers that pass do not appear at all.
-func TestBuildHTFIFOCapabilityRejection_ShapesOperatorMessage(t *testing.T) {
+// TestFormatHTFIFOCapabilityReportForLog_ShapesServerSideDetail
+// pins the SERVER-SIDE log helper's shape — never returned to the
+// client (CodeRabbit major review on PR #734: peer addresses + raw
+// poller errors leak cluster topology to authenticated callers).
+// Each failing peer must contribute a "(reason)" suffix so an
+// operator triaging a rolling-upgrade cluster can fix the lag from
+// the log lines without rerunning the poll out-of-band; peers that
+// pass do not appear.
+func TestFormatHTFIFOCapabilityReportForLog_ShapesServerSideDetail(t *testing.T) {
 	t.Parallel()
 
 	report := &HTFIFOCapabilityReport{
@@ -189,15 +195,41 @@ func TestBuildHTFIFOCapabilityRejection_ShapesOperatorMessage(t *testing.T) {
 		},
 	}
 
-	msg := buildHTFIFOCapabilityRejection(report)
-	require.Contains(t, msg, "every cluster peer to advertise the htfifo capability")
-	require.NotContains(t, msg, "ok:9000", "advertising peers must NOT appear in the rejection")
-	require.Contains(t, msg, "old:9000 (missing capability)")
-	require.Contains(t, msg, "down:9000 (dial tcp: refused)")
+	detail := formatHTFIFOCapabilityReportForLog(report)
+	require.NotContains(t, detail, "ok:9000", "advertising peers must NOT appear in the log detail")
+	require.Contains(t, detail, "old:9000 (missing capability)")
+	require.Contains(t, detail, "down:9000 (dial tcp: refused)")
 
 	// Defensive: nil report and "all-passing-but-AllAdvertise-false" path.
-	require.Contains(t, buildHTFIFOCapabilityRejection(nil), "no report")
+	require.Contains(t, formatHTFIFOCapabilityReportForLog(nil), "no report")
 	allPass := &HTFIFOCapabilityReport{Peers: []HTFIFOCapabilityPeerStatus{{Address: "x", HasHTFIFO: true}}}
-	require.Contains(t, buildHTFIFOCapabilityRejection(allPass), "unknown peer",
-		"never emit a truncated 'did not: ' tail when no peer details surface")
+	require.Contains(t, formatHTFIFOCapabilityReportForLog(allPass), "unknown peer",
+		"never emit an empty detail when no peer reasons surface")
+}
+
+// TestValidateHTFIFOCapability_PublicMessageDoesNotLeakPeerDetails
+// pins the CodeRabbit major review's redaction contract: the
+// client-visible InvalidAttributeValue message MUST NOT include
+// peer addresses or raw poller error text. The two failure-path
+// tests above check that the gate REJECTS; this test specifically
+// checks that the rejection message is the sanitized
+// htfifoCapabilityRejectionPublic constant — no host:port, no raw
+// transport error.
+func TestValidateHTFIFOCapability_PublicMessageDoesNotLeakPeerDetails(t *testing.T) {
+	t.Parallel()
+
+	old := htfifoCapabilityServer(t, []string{}) // no htfifo
+	oldAddr := strings.TrimPrefix(old.URL, "http://")
+	s := &SQSServer{leaderSQS: map[string]string{"raft1": oldAddr}}
+
+	err := s.validateHTFIFOCapability(context.Background(), &sqsQueueMeta{Name: "q.fifo", PartitionCount: 4})
+	require.Error(t, err)
+	var apiErr *sqsAPIError
+	require.True(t, errors.As(err, &apiErr))
+	require.Equal(t, htfifoCapabilityRejectionPublic, apiErr.message,
+		"client message must be the sanitized constant, never the per-peer detail")
+	require.NotContains(t, apiErr.message, oldAddr,
+		"peer host:port MUST NOT appear in the wire-level rejection — operator detail is server-side only")
+	require.Contains(t, apiErr.message, "see server logs for details",
+		"public message must point operators at the server-side detail")
 }

adapter/sqs_catalog.go

@@ -481,9 +481,10 @@ var sqsAttributeAppliers = map[string]attributeApplier{
 	// docs/design/2026_04_26_proposed_sqs_split_queue_fifo.md). Set
 	// at CreateQueue time; SetQueueAttributes attempts to change it
 	// reject via the immutability check in trySetQueueAttributesOnce.
-	// PR 2 of the rollout introduces the field but the temporary
-	// dormancy gate in tryCreateQueueOnce rejects PartitionCount > 1
-	// until PR 5 lifts the gate atomically with the data plane.
+	// PartitionCount > 1 is gated by validateHTFIFOCapability (the
+	// cluster-wide htfifo capability poll, PR 5b-3) which runs
+	// inside tryCreateQueueOnce after the existence check so
+	// idempotent retries do not pay the network cost.
 	"PartitionCount": func(m *sqsQueueMeta, v string) error {
 		// Parse at uint64 width and bound-check explicitly so the
 		// uint32 narrowing below is gosec-clean.
@@ -903,17 +904,12 @@ func (s *SQSServer) createQueueCore(ctx context.Context, in *sqsCreateQueueInput
 	if err != nil {
 		return "", err
 	}
-	// Cluster-wide htfifo capability gate (Phase 3.D §11 PR 5b-3,
-	// replaces the PR 2 dormancy reject). PartitionCount > 1 is
-	// rejected unless this binary AND every peer in s.leaderSQS
-	// advertise the htfifo capability via /sqs_health. The gate
-	// fails closed on any peer timeout, HTTP error, malformed body,
-	// or missing capability so a partitioned queue cannot land in a
-	// partially-upgraded cluster where some peer would silently
-	// store its records under the legacy single-partition keyspace.
-	if err := s.validateHTFIFOCapability(ctx, requested); err != nil {
-		return "", err
-	}
+	// The cluster-wide htfifo capability gate (Phase 3.D §11 PR 5b-3)
+	// runs INSIDE tryCreateQueueOnce after the existence check, not
+	// here. Idempotent CreateQueue retries on an already-existing
+	// partitioned queue must NOT pay the network poll cost or risk a
+	// transient peer-poll failure flipping a 200-OK retry into a 400
+	// (Codex P1 review on PR #734).
 	if len(in.Tags) > sqsMaxTagsPerQueue {
 		// AWS caps tags per queue at 50. CreateQueue must reject
 		// over-cap tag bundles up front; a silent slice-and-store
@@ -963,6 +959,23 @@ func (s *SQSServer) tryCreateQueueOnce(ctx context.Context, requested *sqsQueueM
 		}
 		return false, newSQSAPIError(http.StatusBadRequest, sqsErrQueueNameExists, "queue already exists with different attributes")
 	}
+	// Cluster-wide htfifo capability gate (Phase 3.D §11 PR 5b-3,
+	// replaces the PR 2 dormancy reject). PartitionCount > 1 is
+	// rejected unless this binary AND every peer in s.leaderSQS
+	// advertise the htfifo capability via /sqs_health. Fails closed
+	// on any peer timeout, HTTP error, malformed body, or missing
+	// capability so a partitioned queue cannot land in a partially-
+	// upgraded cluster.
+	//
+	// Runs AFTER the existence check (so idempotent CreateQueue
+	// retries on an already-existing partitioned queue do not pay
+	// the network poll cost or risk a transient peer-poll failure
+	// flipping a 200-OK retry into a 400 — Codex P1 review on PR
+	// #734) and BEFORE the dispatch (so a rejected create does not
+	// burn an OCC commit).
+	if err := s.validateHTFIFOCapability(ctx, requested); err != nil {
+		return false, err
+	}
 	lastGen, err := s.loadQueueGenerationAt(ctx, requested.Name, readTS)
 	if err != nil {
 		return false, errors.WithStack(err)

adapter/sqs_partitioning_integration_test.go

@@ -4,6 +4,8 @@ import (
 	"net/http"
 	"strings"
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 // TestSQSServer_HTFIFO_CapabilityGate_AcceptsOnSingleNode pins the
@@ -35,6 +37,74 @@ func TestSQSServer_HTFIFO_CapabilityGate_AcceptsOnSingleNode(t *testing.T) {
 	}
 }
 
+// TestSQSServer_HTFIFO_CapabilityGate_IsIdempotentOnExistingQueue
+// pins the Codex P1 review fix on PR #734: a CreateQueue retry on
+// an already-existing partitioned queue with identical attributes
+// MUST return 200 (idempotent) even when the cluster-wide
+// capability poll would now fail. Before the fix, the gate ran
+// before the existence check, so a transient peer outage during
+// a CreateQueue retry would flip a 200-OK idempotent response
+// into a 400. Now the gate runs INSIDE tryCreateQueueOnce after
+// the existence check; an existing queue with matching attrs
+// short-circuits to 200 and never touches the network.
+//
+// The test creates a partitioned queue on a single-node cluster
+// (gate passes vacuously), then poisons the SQSServer's peer
+// map with an unreachable address so any subsequent gate
+// invocation would reject. The second CreateQueue with identical
+// attrs must still succeed; only an actually-new partitioned
+// queue would now fail the gate.
+func TestSQSServer_HTFIFO_CapabilityGate_IsIdempotentOnExistingQueue(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	attrs := map[string]string{
+		"FifoQueue":      "true",
+		"PartitionCount": "4",
+	}
+
+	// First create — succeeds because the leaderSQS map is empty
+	// and the local binary advertises htfifo (vacuous gate).
+	status, out := callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName":  "htfifo-idempotent.fifo",
+		"Attributes": attrs,
+	})
+	require.Equal(t, http.StatusOK, status,
+		"first create on single-node cluster must succeed: body=%v", out)
+
+	// Poison the peer map with an unreachable address. Any gate
+	// invocation from this point would call PollSQSHTFIFOCapability
+	// against this dead peer and fail closed → 400. The
+	// post-fix code path skips the gate on an existing-queue
+	// match, so the next create must STILL succeed.
+	require.NotNil(t, node.sqsServer)
+	node.sqsServer.leaderSQS = map[string]string{
+		"raft-fake": "127.0.0.1:1", // unreachable: connection refused
+	}
+
+	status, out = callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName":  "htfifo-idempotent.fifo",
+		"Attributes": attrs,
+	})
+	require.Equal(t, http.StatusOK, status,
+		"second create with identical attrs must be idempotent (gate must NOT run on existing-queue match) — Codex P1 PR #734; body=%v", out)
+
+	// Sanity: the gate IS still in effect for genuinely new
+	// partitioned queues — a different name with the poisoned
+	// peer map must fail the gate.
+	status, out = callSQS(t, node, sqsCreateQueueTarget, map[string]any{
+		"QueueName":  "htfifo-newqueue.fifo",
+		"Attributes": attrs,
+	})
+	require.Equal(t, http.StatusBadRequest, status,
+		"new partitioned queue must hit the gate when a peer is unreachable: body=%v", out)
+	if got, _ := out["__type"].(string); got != sqsErrInvalidAttributeValue {
+		t.Fatalf("new queue: __type=%q (expected InvalidAttributeValue)", got)
+	}
+}
+
 // TestSQSServer_HTFIFO_CapabilityGate_AllowsPartitionCountOne pins
 // the no-op-partition-count path: PartitionCount=1 is the legacy
 // single-partition layout and bypasses the capability gate