fix(adapter/sqs): standard SendMessageBatch double-send under leader churn (#923)

## Summary

Fixes a **double-send** bug in the SQS standard-queue `SendMessageBatch`
path — the same `:duplicate-elements` class fixed for DynamoDB in PR
#920, found by an adversarial S3/SQS audit prompted by the "extend dedup
to S3/SQS" follow-up.

## The bug

`sendMessageBatchWithRetry` → `trySendMessageBatchOnce` →
`sendBatchStandardOnce` runs an **in-process OCC retry loop** that, on
every attempt, re-ran `buildBatchSendRecord` → `buildSendRecord` →
`newMessageIDHex()`, minting a **fresh random MessageID (and send
timestamp) per entry per attempt**, and derived the `data`/`vis`/`byage`
storage keys from them.

Under a Raft leader-election storm, attempt 1 can **commit** its writes
yet surface a self-inflicted `WriteConflict` to the adapter (the
mechanism PR #920 documented). The retry then mints **new** MessageIDs
and writes a **second copy of every entry** at new keys:

1. Attempt 1: mints `{M1,M2,M3}`, commits, but returns `WriteConflict`
under churn.
2. Retry: mints `{M1',M2',M3'}`, commits cleanly, returns
`{M1',M2',M3'}` + HTTP 200.
3. Queue holds **both** sets → every batch entry duplicated; the client
only learns the second set.

Standard queues carry no dedup identity, and the OCC `ReadKeys` are only
`[meta, gen]` (the random data keys never self-conflict), so nothing
fenced this.

**Why only batch:** single `SendMessage` has **no** in-process retry (a
conflict becomes a normal at-least-once SDK retry), and FIFO send is
fenced by the `MessageDeduplicationId` dedup record. The batch path
added an in-process retry without either fence.

## The fix

Pre-generate **one stable `sqsSendIdentity`** (MessageID + receipt token
+ send timestamp) **per entry, before the retry loop**, and reuse it on
every attempt (`buildSendRecordWithIdentity`). The storage keys then
stay constant across retries, so a committed-but-conflicted attempt is
**overwritten idempotently** instead of duplicated.

No FSM probe / `PrevCommitTS` / gate is needed — unlike DynamoDB's
`list_append` (which re-read a *growing* value), the batch write is a
whole-value PUT at a now-stable key, so re-applying is a plain
idempotent overwrite. The record's `QueueGeneration` is still re-derived
per attempt, so a concurrent `DeleteQueue`/`PurgeQueue` generation bump
is followed correctly (the stale-gen attempt-1 write is
orphaned/unreachable).

## S3/SQS audit outcome (recorded in the design doc)

A handler-by-handler adversarial sweep found the original "S3/SQS share
the same latent gap" note was over-broad:
- **S3 — safe.** Whole-value PUT at a stable key; `PutObject` is
OCC-fenced by a caller-supplied `StartTS` (a leader-move retry that
races an interleaved write **fails the fence rather than clobbering**),
and has no in-process re-read-recompute. `CompleteMultipartUpload`'s
manifest is re-assembled deterministically from the **immutable**
uploaded parts (computed once before its retry loop), not from a re-read
of the object value → no duplicate.
- **SQS single `SendMessage` / FIFO — safe** (random-UUID key + no
in-process retry; FIFO dedup record).
- **SQS standard `SendMessageBatch` — fixed here.**

## Tests (`adapter/sqs_batch_send_dedup_test.go`)

- `TestSendMessageBatchStandard_RetryReusesStableKeys` — first dispatch
commits-then-conflicts; asserts the retry dispatches the **same**
storage keys (fresh MessageIDs would fail it → reproduces the bug
pre-fix).
- `TestSendMessageBatchStandard_ReturnedMessageIdsStableAcrossRetry` —
the returned MessageId is the one actually written on both attempts
(client id == stored id).

## Validation

- `go test ./adapter/ -run 'SendMessage|Batch|FIFO|Send'` pass; new
dedup tests pass.
- `golangci-lint run ./adapter/...` 0 issues; `go build ./...` clean.

## Self-review (5 passes)

1. **Data loss** — the fix prevents a double-send (extra messages), and
keeps the generation re-derivation so a concurrent queue delete/recreate
still routes to the live generation; no committed message is dropped.
2. **Concurrency / distributed** — stable keys make the leader-churn
retry an idempotent overwrite; the `[meta,gen]` ReadKeys fence against
`DeleteQueue`/`PurgeQueue` is unchanged.
3. **Performance** — identities are minted once instead of
once-per-attempt (strictly fewer `crypto/rand` calls on the retry path);
hot path (no retry) is one mint per entry, same as before.
4. **Data consistency** — eliminates the duplicate-element anomaly for
standard batch send; FIFO/exactly-once and single-send semantics
unchanged.
5. **Test coverage** — new co-located tests pin key-stability across the
retry (the regression), the load-bearing property of the fix.

Author: bootjp

adapter/sqs_batch_send_dedup_test.go

@@ -0,0 +1,180 @@
+package adapter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+)
+
+// recordingBatchCoordinator records the mutation keys of each dispatch and
+// fails the FIRST dispatch with a retryable WriteConflict — modelling a
+// committed-but-conflicted attempt under leader churn (the apply may have
+// landed, but the dispatch surfaced WriteConflict). The second dispatch
+// succeeds. It embeds stubAdapterCoordinator for the Coordinator surface
+// (IsLeader, Clock, ...) and only overrides Dispatch.
+type recordingBatchCoordinator struct {
+	stubAdapterCoordinator
+	dispatchKeys [][]string
+	// beforeDispatch, if set, runs at the start of each Dispatch with the
+	// 1-based dispatch number — lets a test mutate queue state between attempts.
+	beforeDispatch func(n int)
+}
+
+func (c *recordingBatchCoordinator) Dispatch(_ context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if c.beforeDispatch != nil {
+		c.beforeDispatch(len(c.dispatchKeys) + 1)
+	}
+	keys := make([]string, 0, len(req.Elems))
+	for _, e := range req.Elems {
+		keys = append(keys, string(e.Key))
+	}
+	c.dispatchKeys = append(c.dispatchKeys, keys)
+	if len(c.dispatchKeys) == 1 {
+		return nil, store.ErrWriteConflict
+	}
+	return &kv.CoordinateResponse{}, nil
+}
+
+func seedStandardQueue(t *testing.T, st store.MVCCStore) {
+	const name = "q"
+	t.Helper()
+	meta := &sqsQueueMeta{
+		Name:               name,
+		Generation:         1,
+		MaximumMessageSize: 262144,
+		IsFIFO:             false,
+	}
+	body, err := encodeSQSQueueMeta(meta)
+	require.NoError(t, err)
+	require.NoError(t, st.PutAt(context.Background(), sqsQueueMetaKey(name), body, 1, 0))
+}
+
+// TestSendMessageBatchStandard_RetryReusesStableKeys pins the standard-queue
+// batch-send dedup fix: when the first dispatch commits-then-conflicts under
+// leader churn, the retry must reuse the SAME storage keys (stable MessageIDs +
+// timestamps minted once before the loop) so the committed attempt is
+// overwritten idempotently. Without the fix the retry re-mints fresh
+// MessageIDs, writing a SECOND copy of every entry at new keys — the
+// :duplicate-elements double-send (same class as DynamoDB PR #920).
+func TestSendMessageBatchStandard_RetryReusesStableKeys(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	seedStandardQueue(t, st)
+
+	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
+	srv := &SQSServer{store: st, coordinator: coord}
+
+	entries := []sqsSendMessageBatchEntryInput{
+		{Id: "a", MessageBody: "alpha"},
+		{Id: "b", MessageBody: "bravo"},
+	}
+	successful, failed, err := srv.sendMessageBatchWithRetry(ctx, "q", entries)
+	require.NoError(t, err)
+	require.Empty(t, failed)
+	require.Len(t, successful, 2)
+
+	require.Len(t, coord.dispatchKeys, 2, "attempt 1 conflicts (committed-but-conflicted), attempt 2 retries")
+	require.NotEmpty(t, coord.dispatchKeys[0])
+	require.Equal(t, coord.dispatchKeys[0], coord.dispatchKeys[1],
+		"the retry MUST reuse the same storage keys; fresh MessageIDs would double-send every entry under leader churn")
+}
+
+// TestSendMessageBatchStandard_VisKeyStableAcrossDelayChange pins codex P2:
+// if a SetQueueAttributes changes the queue's DelaySeconds between a
+// committed-but-conflicted attempt and its retry, the retry must still write
+// the SAME vis/by-age keys (AvailableAtMillis is captured in the identity at
+// first mint, not recomputed from the changed delay) — otherwise the first
+// attempt's vis index entry is orphaned and can redeliver the message.
+func TestSendMessageBatchStandard_VisKeyStableAcrossDelayChange(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	seedStandardQueue(t, st) // DelaySeconds = 0
+
+	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
+	coord.beforeDispatch = func(n int) {
+		if n != 2 {
+			return
+		}
+		// SetQueueAttributes raises DelaySeconds to 900 before the retry.
+		meta := &sqsQueueMeta{Name: "q", Generation: 1, MaximumMessageSize: 262144, DelaySeconds: 900}
+		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"}} // no per-message DelaySeconds
+	successful, _, err := srv.sendMessageBatchWithRetry(ctx, "q", entries)
+	require.NoError(t, err)
+	require.Len(t, successful, 1)
+
+	require.Len(t, coord.dispatchKeys, 2)
+	require.Equal(t, coord.dispatchKeys[0], coord.dispatchKeys[1],
+		"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
+// MessageId the client never received.
+func TestSendMessageBatchStandard_ReturnedMessageIdsStableAcrossRetry(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	seedStandardQueue(t, st)
+
+	coord := &recordingBatchCoordinator{stubAdapterCoordinator: stubAdapterCoordinator{clock: kv.NewHLC()}}
+	srv := &SQSServer{store: st, coordinator: coord}
+
+	entries := []sqsSendMessageBatchEntryInput{{Id: "a", MessageBody: "alpha"}}
+	successful, _, err := srv.sendMessageBatchWithRetry(ctx, "q", entries)
+	require.NoError(t, err)
+	require.Len(t, successful, 1)
+
+	// The data key embeds the MessageID; both attempts must carry the
+	// returned MessageID so the client's id matches what is stored.
+	wantDataKey := string(sqsMsgDataKeyDispatch(&sqsQueueMeta{Name: "q", Generation: 1}, "q", 0, 1, successful[0].MessageId))
+	for i, keys := range coord.dispatchKeys {
+		require.Contains(t, keys, wantDataKey, "dispatch %d must write the returned MessageID's data key", i)
+	}
+}

adapter/sqs_messages.go

@@ -653,28 +653,78 @@ func resolveSendDelay(meta *sqsQueueMeta, requested *int64) (int64, error) {
 	return *requested, nil
 }
 
-func buildSendRecord(meta *sqsQueueMeta, in sqsSendMessageInput, delay int64) (*sqsMessageRecord, []byte, error) {
+// sqsSendIdentity is the per-message identity that MUST stay stable across a
+// retry loop. Every field feeds the message's storage keys: the data key via
+// MessageID; the vis key via AvailableAtMillis; the by-age key via
+// SendTimestampMillis. If they were re-minted on every retry attempt, a
+// committed-but-conflicted attempt under leader churn plus the recomputed retry
+// would land at two different key sets, double-sending the message (the
+// :duplicate-elements class fixed for DynamoDB in PR #920).
+//
+// AvailableAtMillis (= sendTs + resolved delay) is captured here too, NOT
+// recomputed per attempt: otherwise a SetQueueAttributes that changes the
+// queue's DelaySeconds between a committed-but-conflicted attempt and its retry
+// would shift the vis key and leave the first attempt's vis entry orphaned
+// (codex P2 on PR #923) — a stale visibility index entry that can redeliver the
+// message. The batch send path lazily mints one identity per entry on its first
+// standard-path attempt and reuses it on every retry, so the keys are stable.
+type sqsSendIdentity struct {
+	messageID         string
+	token             []byte
+	sendTsMillis      int64
+	availableAtMillis int64
+}
+
+// newSendIdentity mints a fresh stable identity for the given resolved delay.
+// Single-message sends call this inline via buildSendRecord (they have no
+// in-process retry, so a self-inflicted conflict surfaces to the client as a
+// normal at-least-once SDK retry); the batch path mints once per entry and
+// reuses across its retry loop.
+func newSendIdentity(delay int64) (sqsSendIdentity, error) {
 	messageID, err := newMessageIDHex()
 	if err != nil {
-		return nil, nil, errors.WithStack(err)
+		return sqsSendIdentity{}, errors.WithStack(err)
 	}
 	token, err := newReceiptToken()
 	if err != nil {
-		return nil, nil, errors.WithStack(err)
+		return sqsSendIdentity{}, errors.WithStack(err)
 	}
 	now := time.Now().UnixMilli()
-	availableAt := now + delay*sqsMillisPerSecond
+	return sqsSendIdentity{
+		messageID:         messageID,
+		token:             token,
+		sendTsMillis:      now,
+		availableAtMillis: now + delay*sqsMillisPerSecond,
+	}, nil
+}
+
+func buildSendRecord(meta *sqsQueueMeta, in sqsSendMessageInput, delay int64) (*sqsMessageRecord, []byte, error) {
+	id, err := newSendIdentity(delay)
+	if err != nil {
+		return nil, nil, err
+	}
+	return buildSendRecordWithIdentity(meta, in, id)
+}
+
+// buildSendRecordWithIdentity builds the message record from a caller-supplied
+// stable identity (MessageID + token + timestamps) instead of minting a fresh
+// one. The record value embeds the current meta.Generation, so callers re-invoke
+// this per retry attempt with the freshly-read generation while keeping the
+// identity fixed — the keys stay stable for a given generation (idempotent
+// overwrite under leader churn) and follow the generation if a concurrent
+// DeleteQueue/PurgeQueue bumps it.
+func buildSendRecordWithIdentity(meta *sqsQueueMeta, in sqsSendMessageInput, id sqsSendIdentity) (*sqsMessageRecord, []byte, error) {
 	body := []byte(in.MessageBody)
 	rec := &sqsMessageRecord{
-		MessageID:              messageID,
+		MessageID:              id.messageID,
 		Body:                   body,
 		MD5OfBody:              sqsMD5Hex(body),
 		MD5OfMessageAttributes: md5OfAttributesHex(in.MessageAttributes),
 		MessageAttributes:      in.MessageAttributes,
-		SendTimestampMillis:    now,
-		AvailableAtMillis:      availableAt,
-		VisibleAtMillis:        availableAt,
-		CurrentReceiptToken:    token,
+		SendTimestampMillis:    id.sendTsMillis,
+		AvailableAtMillis:      id.availableAtMillis,
+		VisibleAtMillis:        id.availableAtMillis,
+		CurrentReceiptToken:    id.token,
 		QueueGeneration:        meta.Generation,
 		MessageGroupId:         in.MessageGroupId,
 		MessageDeduplicationId: in.MessageDeduplicationId,