feat(sqs): Jepsen HT-FIFO workload (Phase 3.D PR 7b) (#738)

## Summary

Phase 3.D PR 7b — Jepsen HT-FIFO workload that stresses partitioned-FIFO
queues against the three contracts AWS HT-FIFO is supposed to honour
even under partition and node-loss faults: **within-group ordering**,
**no message loss**, **no duplicates**.

Pattern follows [aphyr's Jepsen RabbitMQ
analysis](https://aphyr.com/posts/315-jepsen-rabbitmq): track every
`:send` and `:recv` in the operation history, then a custom checker
verifies the contracts against the recorded events at the end of the
run.

## What's in this PR

- **`jepsen/project.clj`** — Adds `com.cognitect.aws/sqs` at the same
version as the existing dynamodb dep, so the SDK wire protocol (auth,
retry classification, error parsing) is exercised end-to-end against
elastickv rather than a hand-rolled HTTP layer.
- **`jepsen/src/elastickv/db.clj`** — Extends `start-node!` to accept
`:sqs-port` (port spec like `:dynamo-port`) and `:sqs-region`. Both are
optional, so existing dynamodb / s3 / redis test specs are
byte-identical at the args level when `sqs-port` is absent.
- **`jepsen/src/elastickv/jepsen_test.clj`** — Registers
`elastickv-sqs-htfifo-test` alongside the other workloads.
- **`jepsen/src/elastickv/sqs_htfifo_workload.clj`** (new, ~430 lines) —
The workload. Uses cognitect/aws-api SQS, creates an HT-FIFO queue with
`PartitionCount=4` + `ContentBasedDeduplication`, runs sends and
receives across N `MessageGroupId` values, and the custom
`ht-fifo-checker` validates the three contracts.
- **`jepsen/test/elastickv/sqs_htfifo_workload_test.clj`** (new) —
Pure-function tests for the checker plus integration smoke tests for the
test-spec builder. 11 tests / 27 assertions.

## Checker contracts

For each `MessageGroupId` independently:

1. **Within-group ordering** — the sequence of received `seq` values,
sorted by global completion time across all consumers, is monotonically
non-decreasing.
2. **No loss** — every `(group, seq)` successfully `:sent` eventually
appears in the `:recv` history. Sends with `:info` status are treated as
possibly-committed and not counted as lost.
3. **No duplicates** — every `(group, seq)` appears at most once in the
`:recv` history. `ContentBasedDeduplication` on the queue + a unique
`(group, seq)` body is what enforces this server-side; a duplicate here
is a real bug (e.g. a deletion that did not commit).

## Open-endpoint mode

The elastickv server starts without `--sqsCredentialsFile`, so the SQS
adapter accepts any signed request (mirroring how the S3 adapter is
wired in jepsen today). The SDK client signs with dummy credentials, so
the SigV4 path still exercises end-to-end at the protocol level.

## Self-review (5 lenses)

1. **Data loss** — N/A; this is a test-only PR. The workload's whole
purpose is to *detect* data loss in the system under test.
2. **Concurrency** — The shared per-group `seq-counter` is an `atom`
updated via `swap!` (CAS-based), so concurrent sends from different
worker threads always assign distinct seqs. The checker is pure; no
shared mutable state.
3. **Performance** — Test-only code, runs at low rate (5
ops/sec/worker). Not on any hot path.
4. **Data consistency** — The checker compares committed sends against
the receive history globally, so all the consistency assertions are at
end-of-run with a complete picture. Sends with `:info` (uncertain
commit) are correctly excluded from the loss set, matching Jepsen's
standard approach.
5. **Test coverage** — 11 unit tests for the checker pin the contract
surface (clean / loss / info-not-loss / duplicates / within-group
ordering / cross-group interleaving / failed-send-not-counted /
empty-receive). Integration smoke tests pin the test-spec builder. The
workload itself is exercised end-to-end on a real cluster via `lein run
-m elastickv.sqs-htfifo-workload`.

## Test plan

- [x] `lein test elastickv.sqs-htfifo-workload-test` — 11 tests / 27
assertions pass
- [x] `lein test` for non-redis suite (dynamodb / dynamodb-types / s3 /
cli / sqs-htfifo) — 21 tests / 41 assertions pass
- [ ] End-to-end live cluster run — operator-driven (out of scope for
the merge gate; relies on a 3-node cluster setup)

The `elastickv.redis-workload` namespace fails to load due to the empty
`redis/src/` tree, which is pre-existing on main and unrelated to this
PR.

## Out of scope (next milestones)

- Wiring the workload into `scripts/run-jepsen-local.sh` — the existing
script is dynamodb-only; an sqs counterpart lands as a follow-up.
- Multi-shard cluster topology that lands distinct partitions on
distinct Raft groups. This PR's `PartitionCount=4` routes to the default
group on a single-shard cluster — partitioning logic (different keys per
partition, ordering preserved within group) is fully exercised, but the
cross-shard scaling story is gated on separate work.
- Design-doc lifecycle rename (`*_proposed_*.md` → `*_partial_*.md`) —
that is §11 PR 8 in the design doc and is tracked separately.

## Refs

- `docs/design/2026_04_26_proposed_sqs_split_queue_fifo.md` §11 PR 7.
- Closes the testing half of §11 PR 7. PR 7a (metrics) shipped at #737.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Added AWS SQS integration with HT‑FIFO support, SQS port/region
configuration, and runtime options to exercise FIFO dedupe/order
semantics

* **Tests**
* Added comprehensive unit and workload tests validating ordering,
no‑loss, no‑duplicates, and option handling

* **Chores**
* CI updated to run the SQS HT‑FIFO workload as part of Jepsen test runs
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

.github/workflows/jepsen-test.yml

@@ -49,6 +49,7 @@ jobs:
           RAFT_REDIS_MAP="127.0.0.1:50051=127.0.0.1:63791,127.0.0.1:50052=127.0.0.1:63792,127.0.0.1:50053=127.0.0.1:63793"
           RAFT_S3_MAP="127.0.0.1:50051=127.0.0.1:63901,127.0.0.1:50052=127.0.0.1:63902,127.0.0.1:50053=127.0.0.1:63903"
           RAFT_DYNAMO_MAP="127.0.0.1:50051=127.0.0.1:63801,127.0.0.1:50052=127.0.0.1:63802,127.0.0.1:50053=127.0.0.1:63803"
+          RAFT_SQS_MAP="127.0.0.1:50051=127.0.0.1:63501,127.0.0.1:50052=127.0.0.1:63502,127.0.0.1:50053=127.0.0.1:63503"
 
           : > /tmp/elastickv-demo.pid
           for node in 1 2 3; do
@@ -57,6 +58,7 @@ jobs:
               --redisAddress "127.0.0.1:6379${node}" \
               --dynamoAddress "127.0.0.1:6380${node}" \
               --s3Address "127.0.0.1:6390${node}" \
+              --sqsAddress "127.0.0.1:6350${node}" \
               --metricsAddress "" \
               --pprofAddress "" \
               --raftId "n${node}" \
@@ -65,15 +67,17 @@ jobs:
               --raftRedisMap "$RAFT_REDIS_MAP" \
               --raftS3Map "$RAFT_S3_MAP" \
               --raftDynamoMap "$RAFT_DYNAMO_MAP" \
+              --raftSqsMap "$RAFT_SQS_MAP" \
               > "/tmp/elastickv-demo-n${node}.log" 2>&1 &
             echo $! >> /tmp/elastickv-demo.pid
           done
 
-          echo "Waiting for redis (63791-63793), dynamo (63801-63803), and s3 (63901-63903) listeners..."
+          echo "Waiting for redis (63791-63793), dynamo (63801-63803), s3 (63901-63903), and sqs (63501-63503) listeners..."
           for i in {1..90}; do
             if nc -z 127.0.0.1 63791 && nc -z 127.0.0.1 63792 && nc -z 127.0.0.1 63793 \
               && nc -z 127.0.0.1 63801 && nc -z 127.0.0.1 63802 && nc -z 127.0.0.1 63803 \
-              && nc -z 127.0.0.1 63901 && nc -z 127.0.0.1 63902 && nc -z 127.0.0.1 63903; then
+              && nc -z 127.0.0.1 63901 && nc -z 127.0.0.1 63902 && nc -z 127.0.0.1 63903 \
+              && nc -z 127.0.0.1 63501 && nc -z 127.0.0.1 63502 && nc -z 127.0.0.1 63503; then
               echo "Cluster is up"
               exit 0
             fi
@@ -142,6 +146,26 @@ jobs:
         timeout-minutes: 3
         run: |
           timeout 120 ~/lein run -m elastickv.s3-workload --local --time-limit 5 --rate 10 --concurrency 10 --s3-ports 63901,63902,63903 --host 127.0.0.1
+      - name: Run SQS HT-FIFO Jepsen workload against elastickv
+        working-directory: jepsen
+        # The HT-FIFO workload runs sends and receives across a 4-partition
+        # FIFO queue with content-based deduplication. The custom checker
+        # validates within-group ordering, no loss, and no duplicates.
+        # See jepsen/src/elastickv/sqs_htfifo_workload.clj.
+        #
+        # --drain-time 15: in --local mode the nemesis is a no-op, so no
+        # message can become invisible due to partition/kill — the 40s
+        # default drain (which protects against fault-induced
+        # visibility-timeout races) is overkill here. 15s leaves ample
+        # headroom under the 120s shell timeout against JVM startup and
+        # the 5s main phase.
+        timeout-minutes: 3
+        run: |
+          timeout 120 ~/lein run -m elastickv.sqs-htfifo-workload --local \
+            --time-limit 5 --rate 5 --concurrency 5 \
+            --partition-count 4 --group-count 6 \
+            --drain-time 15 \
+            --sqs-ports 63501,63502,63503 --host 127.0.0.1
       - name: Stop demo cluster
         if: always()
         run: |

.github/workflows/jepsen.yml

@@ -84,6 +84,22 @@ jobs:
               --faults ${{ github.event.inputs['faults'] || github.event.inputs.faults }} \
               --concurrency 10"
 
+      - name: Run SQS HT-FIFO Jepsen workload
+        working-directory: jepsen
+        env:
+          HOME: ${{ github.workspace }}/jepsen/tmp-home
+          LEIN_HOME: ${{ github.workspace }}/jepsen/.lein
+          LEIN_JVM_OPTS: -Duser.home=${{ github.workspace }}/jepsen/tmp-home
+        run: |
+          vagrant ssh ctrl -c "cd ~/elastickv/jepsen && \
+            lein run -m elastickv.sqs-htfifo-workload \
+              --nodes n1,n2,n3,n4,n5 \
+              --time-limit ${{ github.event.inputs['time-limit'] || github.event.inputs.time-limit }} \
+              --rate ${{ github.event.inputs['rate'] || github.event.inputs.rate }} \
+              --faults ${{ github.event.inputs['faults'] || github.event.inputs.faults }} \
+              --partition-count 4 --group-count 8 \
+              --concurrency 8"
+
       - name: Collect Jepsen artifacts
         if: always()
         working-directory: jepsen

jepsen/project.clj

@@ -13,5 +13,6 @@
                  [com.cognitect.aws/api "0.8.692"]
                  [com.cognitect.aws/endpoints "1.1.12.626"]
                  [com.cognitect.aws/dynamodb "847.2.1365.0"]
+                 [com.cognitect.aws/sqs "847.2.1365.0"]
                  [org.slf4j/slf4j-nop "2.0.9"]]
   :main elastickv.jepsen-test)

jepsen/src/elastickv/db.clj

@@ -97,7 +97,7 @@
          (clojure.string/join ","))))
 
 (defn- start-node!
-  [test node {:keys [bootstrap-node grpc-port redis-port dynamo-port s3-port data-dir raft-groups shard-ranges raft-engine]}]
+  [test node {:keys [bootstrap-node grpc-port redis-port dynamo-port s3-port sqs-port sqs-region data-dir raft-groups shard-ranges raft-engine]}]
   (when (and (seq raft-groups)
              (> (count raft-groups) 1)
              (nil? shard-ranges))
@@ -110,6 +110,8 @@
                  (node-addr node (port-for dynamo-port node)))
         s3 (when s3-port
              (node-addr node (port-for s3-port node)))
+        sqs (when sqs-port
+              (node-addr node (port-for sqs-port node)))
         raft-redis-map (build-raft-redis-map (:nodes test) grpc-port redis-port raft-groups)
         bootstrap? (= node bootstrap-node)
         args (cond-> [server-bin
@@ -121,6 +123,8 @@
                       "--raftRedisMap" raft-redis-map]
                dynamo (conj "--dynamoAddress" dynamo)
                s3 (conj "--s3Address" s3)
+               sqs (conj "--sqsAddress" sqs)
+               (and sqs sqs-region) (conj "--sqsRegion" sqs-region)
                (seq raft-groups) (conj "--raftGroups" (build-raft-groups-arg node raft-groups))
                (seq shard-ranges) (conj "--shardRanges" shard-ranges)
                bootstrap? (conj "--raftBootstrap"))]

jepsen/src/elastickv/jepsen_test.clj

@@ -4,6 +4,7 @@
             [elastickv.dynamodb-workload :as dynamodb-workload]
             [elastickv.dynamodb-types-workload :as dynamodb-types-workload]
             [elastickv.s3-workload :as s3-workload]
+            [elastickv.sqs-htfifo-workload :as sqs-htfifo-workload]
             [jepsen.cli :as cli]))
 
 (defn elastickv-test []
@@ -19,6 +20,14 @@
 (defn elastickv-s3-test []
   (s3-workload/elastickv-s3-test {}))
 
+(defn elastickv-sqs-htfifo-test
+  "HT-FIFO Jepsen test (PR 7b). Run via the workload's own -main:
+   `lein run -m elastickv.sqs-htfifo-workload [opts]`. Same pattern
+   as elastickv-dynamodb-test / elastickv-s3-test — each workload
+   exposes its own -main so this -main only dispatches Redis."
+  ([] (elastickv-sqs-htfifo-test {}))
+  ([opts] (sqs-htfifo-workload/elastickv-sqs-htfifo-test opts)))
+
 (defn -main
   [& args]
   (cli/run! (cli/single-test-cmd {:test-fn elastickv-test}) args))