Add AMQP Imposter and refresh rmqapitests

.github/workflows/fuzz.yml

@@ -0,0 +1,67 @@
+name: Fuzz
+
+on:
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  Fuzz:
+    name: ${{ matrix.suite }}
+    runs-on: ubuntu-24.04
+    timeout-minutes: 240
+    strategy:
+      fail-fast: false
+      matrix:
+        suite: [consumer, producer]
+    env:
+      FUZZ_SEED_COUNT: 1000
+      FUZZ_TARGET: integration/rmqapitests/fuzz/${{ matrix.suite }}_fuzztests.py
+      COMPOSE_FILE: dockerfiles/docker-compose.yaml
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/cache@v4
+        with:
+          path: .vcpkg-cache
+          key: vcpkg-${{ github.run_id }}
+          restore-keys: vcpkg-
+
+      - run: docker compose build dev
+
+      - name: Fuzz
+        shell: bash
+        run: |
+          mkdir -p .vcpkg-cache
+          docker compose run --rm --no-deps -T \
+            -v "$GITHUB_WORKSPACE/.vcpkg-cache:/vcpkg-cache" \
+            -e VCPKG_DEFAULT_BINARY_CACHE=/vcpkg-cache \
+            -e FUZZ_SEED_COUNT -e FUZZ_SEED_OFFSET -e FUZZ_TARGET -e COLUMNS=250 \
+            -e FUZZ_TRANSCRIPT_DIR=/workarea/fuzz-transcripts \
+            dev make fuzz 2>&1 | tee fuzz.log
+
+      - uses: actions/upload-artifact@v4
+        if: ${{ failure() }}
+        with:
+          name: fuzz-transcripts-${{ matrix.suite }}
+          path: fuzz-transcripts/
+          if-no-files-found: ignore
+
+      - name: Summarise
+        if: ${{ failure() }}
+        run: |
+          {
+            offset=${FUZZ_SEED_OFFSET:-0}
+            echo "### ${{ matrix.suite }}, seeds $offset to $(( offset + FUZZ_SEED_COUNT - 1 ))"
+            echo '```'
+            grep -E '^(FAILED|ERROR) |^SUMMARY: [A-Za-z]+Sanitizer' fuzz.log || echo "No test results, see the Fuzz step"
+            echo '```'
+          } >> "$GITHUB_STEP_SUMMARY"

Makefile

@@ -8,6 +8,16 @@ CMAKE_PRESET?=macos-arm64-vcpkg
 RMQ_HOSTNAME?=localhost
 RMQ_MGMT?="http://localhost:15672"
 
+IMPOSTER_FOLDER?=$(SRC_FOLDER)/src/tests/amqp_imposter
+IMPOSTER_VENV?=$(BUILD_FOLDER)/imposter-venv
+FUZZ_BUILD_FOLDER?=$(BUILD_FOLDER)/fuzz-asan
+FUZZ_CXX_FLAGS?=-fsanitize=address -fno-omit-frame-pointer
+FUZZ_ARGS?=-n auto
+FUZZ_TARGET?=integration/rmqapitests/fuzz
+FUZZ_SEED_COUNT?=1000
+FUZZ_SEED_OFFSET?=0
+FUZZ_TRANSCRIPT_DIR?=$(BUILD_FOLDER)/fuzz-transcripts
+
 
 default: build unit
 
@@ -22,12 +32,32 @@ unit:
 
 full: init build 
 
-full-test: full unit integration
+full-test: full unit integration imposter-test
 
 integration: 
 	RMQ_USER=guest RMQ_PWD=guest RMQ_USER_MGMT="guest" RMQ_PWD_MGMT="guest" RMQ_PORT="5672" RMQ_TLS_PORT="5671" RMQ_PROXY_CONTROL_PORT="5700" \
 	cmake --build $(BUILD_FOLDER) -t test_integration 
 
+imposter-venv:
+	uv venv --allow-existing --python 3.15 $(IMPOSTER_VENV)
+	uv pip install --python $(IMPOSTER_VENV) -e $(IMPOSTER_FOLDER) -r $(IMPOSTER_FOLDER)/requirements-dev.txt
+
+imposter-test: imposter-venv
+	cd $(IMPOSTER_FOLDER) && $(abspath $(IMPOSTER_VENV))/bin/python -m pytest -n auto
+
+fuzz-build:
+	cmake --preset ${CMAKE_PRESET} -B ${FUZZ_BUILD_FOLDER} -S ${SRC_FOLDER} -G Ninja -DCMAKE_CXX_FLAGS="$(FUZZ_CXX_FLAGS)"
+	cmake --build ${FUZZ_BUILD_FOLDER} -t rmqperftest
+
+fuzz: imposter-venv fuzz-build
+	uv pip install --python $(IMPOSTER_VENV) -r $(SRC_FOLDER)/src/tests/integration/requirements.txt
+	rm -rf $(FUZZ_TRANSCRIPT_DIR)
+	cd $(SRC_FOLDER)/src/tests && \
+	RMQ_EXE=$(abspath $(FUZZ_BUILD_FOLDER))/examples/rmqperftest/rmqperftest FUZZ_SEED_COUNT=$(FUZZ_SEED_COUNT) FUZZ_SEED_OFFSET=$(FUZZ_SEED_OFFSET) \
+	FUZZ_TRANSCRIPT_DIR=$(abspath $(FUZZ_TRANSCRIPT_DIR)) RMQAPITESTS_IMMEDIATE_RETRY=1 PYTHONPATH=. \
+	$(abspath $(IMPOSTER_VENV))/bin/python -m pytest -p integration.rmqapitests.fuzz.fixtures -o python_files='*_fuzztests.py' \
+	--rootdir . --timeout 600 $(FUZZ_ARGS) '$(FUZZ_TARGET)'
+
 clean:
 	rm -rf $(BUILD_FOLDER)
 
@@ -43,4 +73,4 @@ docker-test:
 docker-shell:
 	cd $(SRC_FOLDER)/dockerfiles && docker compose run dev
 
-.PHONY: build init unit integration clean full full-test
+.PHONY: build init unit integration clean full full-test imposter-venv imposter-test fuzz-build fuzz

dockerfiles/dev.Dockerfile

@@ -26,6 +26,9 @@ RUN apt-get update && apt-get install -y \
 
 RUN pip3 install --break-system-packages --no-cache-dir "cmake>=4.3,<5"
 
+RUN pip3 install --break-system-packages --no-cache-dir "uv>=0.12,<0.14" && \
+    uv python install 3.15
+
 ENV VCPKG_FORCE_SYSTEM_BINARIES=1
 
 # clone and install vcpkg

examples/rmqperftest/rmqperftest.m.cpp

@@ -78,6 +78,11 @@ int main(int argc, char* argv[])
         args.shuffleConnectionEndpoints = true;
     }
 
+    char* envImmediateRetry = std::getenv("RMQAPITESTS_IMMEDIATE_RETRY");
+    if (envImmediateRetry && bsl::string(envImmediateRetry) == "1") {
+        args.immediateRetry = true;
+    }
+
     balcl::OptionInfo specTable[] = {
         {
             "d|id",

examples/rmqperftest/rmqperftest_args.h

@@ -105,6 +105,7 @@ struct PerfTestArgs {
     , consumer()
     , producer()
     , shuffleConnectionEndpoints(false)
+    , immediateRetry(false)
     {
     }
 
@@ -118,6 +119,7 @@ struct PerfTestArgs {
     ProducerArgs producer;
 
     bool shuffleConnectionEndpoints;
+    bool immediateRetry;
 };
 
 } // namespace rmqperftest

examples/rmqperftest/rmqperftest_runner.cpp

@@ -161,6 +161,9 @@ int Runner::run(const PerfTestArgs& args)
 {
     rmqa::RabbitContextOptions options;
     options.setShuffleConnectionEndpoints(args.shuffleConnectionEndpoints);
+    if (args.immediateRetry) {
+        options.setTunable("IIR");
+    }
     rmqa::RabbitContext rabbit(options);
 
     bsl::optional<rmqt::VHostInfo> vhostInfo =

src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp

@@ -227,7 +227,8 @@ RabbitContextImpl::RabbitContextImpl(
             options.clientProperties(),
             options.connectionErrorThreshold(),
             options.connectionEstablishmentTimeout(),
-            isHostHealthMonitoringEnabled);
+            isHostHealthMonitoringEnabled,
+            d_tunables);
 
     if (!d_threadPool) {
         bslmt::ThreadAttributes attributes;

src/rmq/rmqamqp/rmqamqp_connection.cpp

@@ -1169,7 +1169,8 @@ Connection::Factory::Factory(
     const rmqt::FieldTable& clientProperties,
     const bsl::optional<bsls::TimeInterval>& connectionErrorThreshold,
     const bsl::optional<bsls::TimeInterval>& connectionEstablishmentTimeout,
-    const bool isHostHealthMonitoringEnabled)
+    const bool isHostHealthMonitoringEnabled,
+    const rmqt::Tunables& tunables)
 : d_errorCb(errorCb)
 , d_clientProperties(clientProperties)
 , d_metricPublisher(metricPublisher)
@@ -1179,6 +1180,7 @@ Connection::Factory::Factory(
 , d_connectionErrorThreshold(connectionErrorThreshold)
 , d_connectionEstablishmentTimeout(connectionEstablishmentTimeout)
 , d_isHostHealthMonitoringEnabled(isHostHealthMonitoringEnabled)
+, d_tunables(tunables)
 {
 }
 
@@ -1207,17 +1209,23 @@ bsl::shared_ptr<Connection> Connection::Factory::create(
 
 bsl::shared_ptr<rmqio::RetryHandler> Connection::Factory::newRetryHandler()
 {
+    const bsl::shared_ptr<rmqio::RetryStrategy> retryStrategy =
+        d_tunables.find("IIR") != d_tunables.end()
+            ? bsl::make_shared<rmqio::BackoffLevelRetryStrategy>(
+                  bsl::numeric_limits<unsigned int>::max())
+            : bsl::make_shared<rmqio::BackoffLevelRetryStrategy>();
+
     return d_connectionErrorThreshold
                ? bsl::shared_ptr<rmqio::RetryHandler>(
                      bsl::make_shared<rmqio::ConnectionRetryHandler>(
                          d_timerFactory,
                          d_errorCb,
-                         bsl::make_shared<rmqio::BackoffLevelRetryStrategy>(),
+                         retryStrategy,
                          *d_connectionErrorThreshold))
                : bsl::make_shared<rmqio::RetryHandler>(
                      d_timerFactory,
                      d_errorCb,
-                     bsl::make_shared<rmqio::BackoffLevelRetryStrategy>());
+                     retryStrategy);
 }
 
 bsl::shared_ptr<rmqamqp::HeartbeatManager>

src/rmq/rmqamqp/rmqamqp_connection.h

@@ -34,6 +34,7 @@
 #include <rmqt_credentials.h>
 #include <rmqt_endpoint.h>
 #include <rmqt_message.h>
+#include <rmqt_properties.h>
 
 #include <ball_log.h>
 #include <bslmt_mutex.h>
@@ -340,7 +341,8 @@ class Connection::Factory {
         const rmqt::FieldTable& clientProperties,
         const bsl::optional<bsls::TimeInterval>& connectionErrorThreshold,
         const bsl::optional<bsls::TimeInterval>& connectionEstablishmentTimeout,
-        const bool isHostHealthMonitoringEnabled);
+        const bool isHostHealthMonitoringEnabled,
+        const rmqt::Tunables& tunables = rmqt::Tunables());
 
     virtual ~Factory() {}
 
@@ -367,6 +369,7 @@ class Connection::Factory {
     const bsl::optional<bsls::TimeInterval> d_connectionErrorThreshold;
     const bsl::optional<bsls::TimeInterval> d_connectionEstablishmentTimeout;
     const bool d_isHostHealthMonitoringEnabled;
+    const rmqt::Tunables d_tunables;
 }; // class Connection::Factory
 } // namespace rmqamqp
 } // namespace BloombergLP

src/tests/amqp_imposter/.gitattributes

@@ -0,0 +1,4 @@
+*_pb2.py linguist-generated=true
+*_pb2.pyi linguist-generated=true
+*_pb2_grpc.py linguist-generated=true
+*_pb2_grpc.pyi linguist-generated=true

src/tests/amqp_imposter/.gitignore

@@ -0,0 +1,27 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# *egg* files, which can be generated with pip install -e .
+eggs/
+.eggs/
+*.egg
+*.egg-info/
+
+
+# Unit test / coverage reports
+.coverage
+coverage.xml
+coverage-*.xml
+junit-*.xml
+tests.xml
+unit-test.xml
+integration-test.xml
+system-test.xml
+
+.mypy_cache/
+.pytest_cache/
+.ruff_cache/
+
+*.lock