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

.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))