fix(adapter/sqs): PR #923 round-2 — re-validate batch entries against current meta every attempt

codex P2 round-2: round-1 moved per-entry validation into the first-mint path,
so an entry whose identity was minted by a prior UNCOMMITTED attempt (lost a
normal OCC race) skipped validation on retry. If a concurrent SetQueueAttributes
lowered MaximumMessageSize (or tightened another meta-dependent limit) in the
meantime, the retry could commit a message the fresh meta should reject.

Split the two concerns:
- validateBatchEntry runs against the CURRENT meta on EVERY attempt (body
  required, MaximumMessageSize, attributes, FIFO params, delay bounds) and
  returns the resolved delay.
- The stable identity (MessageID/token/timestamps incl. AvailableAtMillis) is
  still minted ONCE and reused, so the keys stay stable across retries (round-1
  vis-key stability preserved); AvailableAtMillis is seeded from the first
  attempt's delay, and re-resolving delay per attempt only re-checks bounds.

Caller audit: validateBatchEntry has one caller (buildBatchSendRecord);
buildBatchSendRecord has one caller (sendBatchStandardOnce); mintBatchSendIdentity
removed (no refs). Single-send/FIFO paths unaffected (buildSendRecord unchanged).

Test: TestSendMessageBatchStandard_RevalidatesAgainstCurrentMetaOnRetry — a
mid-flight MaximumMessageSize drop must make the retry reject the now-too-big
entry.

Author: bootjp

adapter/sqs_batch_send_dedup_test.go

@@ -38,7 +38,8 @@ func (c *recordingBatchCoordinator) Dispatch(_ context.Context, req *kv.Operatio
 	return &kv.CoordinateResponse{}, nil
 }
 
-func seedStandardQueue(t *testing.T, st store.MVCCStore, name string) {
+func seedStandardQueue(t *testing.T, st store.MVCCStore) {
+	const name = "q"
 	t.Helper()
 	meta := &sqsQueueMeta{
 		Name:               name,
@@ -62,7 +63,7 @@ func TestSendMessageBatchStandard_RetryReusesStableKeys(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 	st := store.NewMVCCStore()
-	seedStandardQueue(t, st, "q")
+	seedStandardQueue(t, st)
 
 	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
 	srv := &SQSServer{store: st, coordinator: coord}
@@ -92,7 +93,7 @@ func TestSendMessageBatchStandard_VisKeyStableAcrossDelayChange(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 	st := store.NewMVCCStore()
-	seedStandardQueue(t, st, "q") // DelaySeconds = 0
+	seedStandardQueue(t, st) // DelaySeconds = 0
 
 	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
 	coord.beforeDispatch = func(n int) {
@@ -117,6 +118,41 @@ func TestSendMessageBatchStandard_VisKeyStableAcrossDelayChange(t *testing.T) {
 		"vis/by-age keys must stay stable across a mid-retry DelaySeconds change (cached AvailableAtMillis)")
 }
 
+// TestSendMessageBatchStandard_RevalidatesAgainstCurrentMetaOnRetry pins codex
+// P2 round-2: if a prior attempt minted the identity but did NOT commit (lost a
+// normal OCC race) and a concurrent SetQueueAttributes then lowered
+// MaximumMessageSize below the body, the retry must re-validate against the
+// fresh meta and REJECT the entry rather than commit it on the cached identity.
+func TestSendMessageBatchStandard_RevalidatesAgainstCurrentMetaOnRetry(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	seedStandardQueue(t, st) // MaximumMessageSize = 262144
+
+	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
+	coord.beforeDispatch = func(n int) {
+		if n != 1 {
+			return
+		}
+		// SetQueueAttributes lowers MaximumMessageSize below the body size,
+		// landing during attempt 1's dispatch (which does not commit — it
+		// returns a WriteConflict). The retry then loads this lowered meta
+		// BEFORE re-validating, so the entry must be rejected.
+		meta := &sqsQueueMeta{Name: "q", Generation: 1, MaximumMessageSize: 3}
+		body, err := encodeSQSQueueMeta(meta)
+		require.NoError(t, err)
+		require.NoError(t, st.PutAt(ctx, sqsQueueMetaKey("q"), body, 100, 0))
+	}
+	srv := &SQSServer{store: st, coordinator: coord}
+
+	entries := []sqsSendMessageBatchEntryInput{{Id: "a", MessageBody: "alpha"}} // 5 bytes > new limit 3
+	successful, failed, err := srv.sendMessageBatchWithRetry(ctx, "q", entries)
+	require.NoError(t, err)
+	require.Empty(t, successful, "the retry must re-validate against the lowered limit and reject the entry")
+	require.Len(t, failed, 1)
+	require.Equal(t, "a", failed[0].Id)
+}
+
 // TestSendMessageBatchStandard_ReturnedMessageIdsStableAcrossRetry pins that
 // the MessageIds reported to the client are the ones actually stored on the
 // retry (not a discarded first-attempt set), so a consumer cannot observe a
@@ -125,7 +161,7 @@ func TestSendMessageBatchStandard_ReturnedMessageIdsStableAcrossRetry(t *testing
 	t.Parallel()
 	ctx := context.Background()
 	st := store.NewMVCCStore()
-	seedStandardQueue(t, st, "q")
+	seedStandardQueue(t, st)
 
 	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
 	srv := &SQSServer{store: st, coordinator: coord}

adapter/sqs_messages_batch.go

@@ -407,15 +407,20 @@ func (s *SQSServer) resolveFreshFifoSnapshot(ctx context.Context, queueName stri
 // SendMessage would, but returns the *sqsAPIError so the batch path
 // can drop the entry into Failed[] instead of failing the whole
 // request.
-// buildBatchSendRecord builds one standard-queue batch entry's record, reusing
-// a stable per-entry identity across retries. id points into the caller's
-// identities slice: on the FIRST attempt for this entry (id.messageID == "")
-// it validates, resolves the delay, and mints the identity (capturing
-// AvailableAtMillis from the delay in effect at that moment); on retries it
-// skips validation/resolution and reuses the cached identity, so the storage
-// keys are stable even if a concurrent SetQueueAttributes changes the queue's
-// DelaySeconds between attempts (codex P2). An entry that already committed on a
-// prior attempt is never re-validated against a since-changed meta.
+// buildBatchSendRecord builds one standard-queue batch entry's record. It splits
+// two concerns that must NOT be conflated:
+//
+//   - Validation against the CURRENT meta runs on EVERY attempt (`validateBatchEntry`).
+//     A prior, uncommitted attempt may have minted the identity; if a concurrent
+//     SetQueueAttributes then tightened a limit (e.g. lowered MaximumMessageSize),
+//     the retry must still reject a now-invalid entry rather than commit it
+//     (codex P2 round-2).
+//   - The stable identity (MessageID + token + timestamps incl. AvailableAtMillis)
+//     is minted ONCE, on the first attempt, and reused. This keeps the storage
+//     keys constant across retries so a committed-but-conflicted attempt is
+//     overwritten idempotently, and so a mid-retry DelaySeconds change cannot
+//     shift the vis key (codex P2 round-1). AvailableAtMillis is captured from
+//     the first attempt's resolved delay.
 func buildBatchSendRecord(meta *sqsQueueMeta, entry sqsSendMessageBatchEntryInput, id *sqsSendIdentity) (*sqsMessageRecord, []byte, error) {
 	asSingle := sqsSendMessageInput{
 		MessageBody:            entry.MessageBody,
@@ -424,39 +429,41 @@ func buildBatchSendRecord(meta *sqsQueueMeta, entry sqsSendMessageBatchEntryInpu
 		MessageGroupId:         entry.MessageGroupId,
 		MessageDeduplicationId: entry.MessageDeduplicationId,
 	}
+	delay, err := validateBatchEntry(meta, entry, asSingle)
+	if err != nil {
+		return nil, nil, err
+	}
 	if id.messageID == "" {
-		// First attempt for this entry: validate, resolve delay, mint identity.
-		minted, err := mintBatchSendIdentity(meta, entry, asSingle)
-		if err != nil {
-			return nil, nil, err
+		minted, mintErr := newSendIdentity(delay)
+		if mintErr != nil {
+			return nil, nil, mintErr
 		}
 		*id = minted
 	}
 	return buildSendRecordWithIdentity(meta, asSingle, *id)
 }
 
-// mintBatchSendIdentity validates one batch entry against the current queue
-// meta, resolves its delay, and mints the stable identity. It runs only on the
-// first attempt for an entry; retries reuse the cached identity so a
-// since-changed meta cannot re-fail an already-committed entry or shift its keys.
-func mintBatchSendIdentity(meta *sqsQueueMeta, entry sqsSendMessageBatchEntryInput, asSingle sqsSendMessageInput) (sqsSendIdentity, error) {
+// validateBatchEntry runs every per-attempt check for a standard batch entry
+// against the currently-loaded queue meta and returns the resolved delay. It is
+// invoked on EVERY attempt (not just the first mint) so a concurrent
+// SetQueueAttributes that tightened limits is honored on the retry. The returned
+// delay seeds AvailableAtMillis only on the first mint; on retries the cached
+// identity's AvailableAtMillis is reused, so re-resolving here never shifts the
+// keys — it only re-checks bounds.
+func validateBatchEntry(meta *sqsQueueMeta, entry sqsSendMessageBatchEntryInput, asSingle sqsSendMessageInput) (int64, error) {
 	if len(entry.MessageBody) == 0 {
-		return sqsSendIdentity{}, newSQSAPIError(http.StatusBadRequest, sqsErrValidation, "MessageBody is required")
+		return 0, newSQSAPIError(http.StatusBadRequest, sqsErrValidation, "MessageBody is required")
 	}
 	if int64(len(entry.MessageBody)) > meta.MaximumMessageSize {
-		return sqsSendIdentity{}, newSQSAPIError(http.StatusBadRequest, sqsErrMessageTooLong, "message body exceeds MaximumMessageSize")
+		return 0, newSQSAPIError(http.StatusBadRequest, sqsErrMessageTooLong, "message body exceeds MaximumMessageSize")
 	}
 	if err := validateMessageAttributes(entry.MessageAttributes); err != nil {
-		return sqsSendIdentity{}, err
+		return 0, err
 	}
 	if err := validateSendFIFOParams(meta, asSingle); err != nil {
-		return sqsSendIdentity{}, err
-	}
-	delay, err := resolveSendDelay(meta, entry.DelaySeconds)
-	if err != nil {
-		return sqsSendIdentity{}, err
+		return 0, err
 	}
-	return newSendIdentity(delay)
+	return resolveSendDelay(meta, entry.DelaySeconds)
 }
 
 // ------------------------ DeleteMessageBatch ------------------------