feat: a new tool, anarkey, to help artificially load test arke

.github/dependabot.yml

@@ -20,6 +20,38 @@ updates:
         versions:
           - "1.7.0"
           - "1.7.1"
+  - package-ecosystem: gomod
+    directory: "/api"
+    schedule:
+      interval: "weekly"
+    groups:
+      gomod:
+        update-types:
+          - minor
+          - patch
+  - package-ecosystem: gomod
+    directory: "/tests/integration"
+    schedule:
+      interval: "weekly"
+    groups:
+      gomod:
+        update-types:
+          - minor
+          - patch
+    ignore:
+      - dependency-name: github.com/rabbitmq/rabbitmq-stream-go-client
+        versions:
+          - "1.7.0"
+          - "1.7.1"
+  - package-ecosystem: gomod
+    directory: "/tools/anarkey"
+    schedule:
+      interval: "weekly"
+    groups:
+      gomod:
+        update-types:
+          - minor
+          - patch
   - package-ecosystem: github-actions
     directory: "/"
     schedule:

.github/workflows/ci.yaml

@@ -62,8 +62,9 @@ jobs:
         if: always()
         uses: ludeeus/action-shellcheck@00b27aa7cb85167568cb48a3838b75f4265f2bca
 
+      # yamllint does not exist in the act image so skip it if we're in act
       - name: Run yamllint
-        if: always()
+        if: ${{ !env.ACT }}
         run: |
           yamllint -c .github/linters/yamllint.yml .
 

.gitignore

@@ -18,3 +18,4 @@ coverage/
 *.out
 gitleaks
 .DS_Store
+tools/anarkey/logs

internal/provider/connectors/amqp091/amqp091.go

@@ -1262,7 +1262,7 @@ func (prov *amqp091provider) streamSubscribe(ctx context.Context, bd *BrokerDeta
 		if err := latch.Increment(); err != nil {
 			return
 		}
-		messageChannel <- m
+
 		stm := streamMessage{Body: message.GetData(), Headers: m.GetHeaders()}
 		stm.Ack = func() {
 			latch.Decrement()
@@ -1280,7 +1280,11 @@ func (prov *amqp091provider) streamSubscribe(ctx context.Context, bd *BrokerDeta
 			util.Logger.Tracef("Nack of message(%s) on stream %s%s with offset %d", messageUUID, ctx.Consumer.GetStreamName(), consumerGroup, ctx.Consumer.GetOffset())
 			latch.Decrement()
 		}
+
+		// add the stream message to active messages before we send it to the client to avoid the client
+		// potentially acking before we add it to active messages (can happen in low latency envs
 		bd.activeMessages.Add(messageUUID, stm)
+		messageChannel <- m
 		atomic.AddInt64(&bd.consumed, 1)
 	}
 
@@ -1933,7 +1937,7 @@ func (bd *BrokerDetails) loadExchanges() {
 		return
 	}
 	if marshErr := json.Unmarshal(body, &results); marshErr != nil {
-		util.Logger.Debugf("Error unmarshaling exchanges from management API: %s", err.Error())
+		util.Logger.Debugf("Error unmarshaling exchanges from management API: %s", marshErr.Error())
 		return
 	}
 	util.Logger.Debugf("Loaded exchanges from management API: %s", string(body))

internal/provider/connectors/amqp091/streamshim.go

@@ -274,7 +274,7 @@ func (sc *streamConnection) GetLastOffset(streamName string, consumerName string
 }
 
 func (sc *streamConnection) StoreOffset(streamName string, consumerName string, offset int64) error {
-	util.Logger.Debugf("StoreOffset (%s)(%s)(%d)", consumerName, streamName, offset)
+	util.Logger.Tracef("StoreOffset (%s)(%s)(%d)", consumerName, streamName, offset)
 	return sc.env.StoreOffset(consumerName, streamName, offset)
 }
 

tests/integration/go.mod

@@ -52,7 +52,7 @@ require (
 	golang.org/x/oauth2 v0.35.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/term v0.41.0 // indirect
-	golang.org/x/text v0.35.0 // indirect
+	golang.org/x/text v0.36.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect

tests/integration/go.sum

@@ -118,8 +118,8 @@ go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
 go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
-golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
-golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
+golang.org/x/mod v0.34.0 h1:xIHgNUUnW6sYkcM5Jleh05DvLOtwc6RitGHbDk4akRI=
+golang.org/x/mod v0.34.0/go.mod h1:ykgH52iCZe79kzLLMhyCUzhMci+nQj+0XkbXpNYtVjY=
 golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
 golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
 golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ=
@@ -131,12 +131,12 @@ golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU=
 golang.org/x/term v0.41.0/go.mod h1:3pfBgksrReYfZ5lvYM0kSO0LIkAl4Yl2bXOkKP7Ec2A=
-golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
-golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
+golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
+golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
-golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
-golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
+golang.org/x/tools v0.43.0 h1:12BdW9CeB3Z+J/I/wj34VMl8X+fEXBxVR90JeMX5E7s=
+golang.org/x/tools v0.43.0/go.mod h1:uHkMso649BX2cZK6+RpuIPXS3ho2hZo4FVwfoy1vIk0=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA=

tools/anarkey/Makefile

@@ -0,0 +1,4 @@
+.PHONY: build
+
+build:
+	go build -o build/anarkey .