From 44626f1beecd8b14dba4782045506ce3900e99f2 Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 5 Nov 2026 11:05:48 +0800 Subject: [PATCH] fix(tier): prepare and verify MQTT dependencies before setup --- .github/workflows/rustfs-tier-test.yml | 42 ++--- README.md | 4 + rustfs-tier-test.sh | 172 +++++++++++++++-- test_rustfs_tier.sh | 2 + test_rustfs_tier_mqtt.py | 245 +++++++++++++++++++++++++ 5 files changed, 427 insertions(+), 38 deletions(-) create mode 100644 test_rustfs_tier_mqtt.py diff --git a/.github/workflows/rustfs-tier-test.yml b/.github/workflows/rustfs-tier-test.yml index 827b9cb..2f1d57a 100644 --- a/.github/workflows/rustfs-tier-test.yml +++ b/.github/workflows/rustfs-tier-test.yml @@ -216,8 +216,6 @@ jobs: timeout-minutes: 5 run: | set -euo pipefail - sudo docker rm -f rustfs-test-mqtt >/dev/null 2>&1 || true - sudo rm -f /tmp/rustfs-mosquitto.conf read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002}" SSH_USER="${RUSTFS_SSH_USER:-azureuser}" for node in "${NODES[@]}"; do @@ -233,34 +231,16 @@ jobs: ' done - - name: Ensure MQTT broker + clients + - name: Ensure MQTT clients timeout-minutes: 5 run: | set -euo pipefail - if ! command -v mosquitto_sub >/dev/null 2>&1; then + if ! command -v mosquitto_sub >/dev/null 2>&1 || ! command -v mosquitto_pub >/dev/null 2>&1; then sudo apt-get update sudo apt-get install -y mosquitto-clients fi - command -v docker >/dev/null 2>&1 || { echo 'docker not found on runner'; exit 1; } - sudo docker rm -f rustfs-test-mqtt >/dev/null 2>&1 || true - cat <<'EOF' | sudo tee /tmp/rustfs-mosquitto.conf >/dev/null - listener 1883 0.0.0.0 - allow_anonymous true - EOF - sudo docker run -d --name rustfs-test-mqtt -p 1883:1883 \ - -v /tmp/rustfs-mosquitto.conf:/mosquitto/config/mosquitto.conf:ro \ - eclipse-mosquitto:2 >/dev/null - for _ in {1..10}; do - if ss -tln 2>/dev/null | grep -q ':1883'; then - break - fi - sleep 1 - done - ss -tln 2>/dev/null | grep -q ':1883' || { - echo 'mosquitto container is not listening on 1883' - sudo docker logs rustfs-test-mqtt || true - exit 1 - } + # The suite owns broker preparation, readiness, logs, and cleanup so + # direct VM invocations exercise the same prerequisite as CI. - name: Run tier suite id: test @@ -467,8 +447,17 @@ jobs: if: ${{ always() && steps.environment.outcome == 'success' }} run: | set -euo pipefail - sudo docker rm -f rustfs-test-mqtt >/dev/null 2>&1 || true - sudo rm -f /tmp/rustfs-mosquitto.conf + MQTT_CLEANUP_STATUS=0 + if [ -s "${TIER_ARTIFACTS_DIR}/mqtt-container-id" ]; then + # Recover only this run's container if cancellation interrupted + # the suite's EXIT handler. Reuse its receipt validation and cleanup. + TIER_SOURCE_ONLY=1 source ./rustfs-tier-test.sh + MQTT_CIDFILE="${TIER_ARTIFACTS_DIR}/mqtt-container-id" + RESULT_LOG_DIR="${TIER_ARTIFACTS_DIR}/logs" + MQTT_DOCKER=(sudo -n docker) + MQTT_BROKER_OWNED=1 + mqtt_broker_stop || MQTT_CLEANUP_STATUS=1 + fi read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002}" SSH_USER="${RUSTFS_SSH_USER:-azureuser}" for node in "${NODES[@]}"; do @@ -483,6 +472,7 @@ jobs: ${SUDO} rm -rf /var/log/rustfs /var/lib/rustfs/kms ' done + exit "${MQTT_CLEANUP_STATUS}" - name: Release environment only after successful cleanup if: ${{ always() && steps.environment.outcome == 'success' }} diff --git a/README.md b/README.md index 04f743a..425f64a 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,10 @@ python3 -B -m unittest -v test_rustfs_security.py test_sensitive_diagnostics.py ## Tier suite artifacts +The full Tier suite requires `mosquitto_sub` and `mosquitto_pub`. Before changing any RustFS node, it checks MQTT publish/subscribe delivery through `${RUSTFS_HEAL_IP}:1883`. An existing broker is reused. If the endpoint is local and has no listener, the suite starts an isolated `eclipse-mosquitto:2` Docker container and removes only that container on exit, including failed setup. Docker must be available directly or through `sudo -n`. Set `RUSTFS_TIER_MQTT_MODE=external` to require an already running broker without creating one; use `RUSTFS_MQTT_PORT` to override port 1883. Remote brokers must be provisioned separately. Canonical tiering, events, and audit validation cases do not require MQTT. + +MQTT startup, subscriber, broker, and VM connectivity diagnostics are stored under the artifact `logs/` directory. A managed container's `mqtt-container-id` receipt remains in the artifact directory until cleanup succeeds; CI uses the same cleanup function to recover that container after cancellation. A failed module-enable request also captures bounded, credential-filtered notification startup logs from the source and target before teardown. The test servers use the `warn` log level so lifecycle failure details are available. Always pass a unique persistent `--artifacts-dir` when investigating a setup failure; the default temporary directory is removed on exit. + `rustfs-tier-test.sh` can persist machine-readable evidence with `--artifacts-dir `. The directory contains one atomic JSON result per `(topology, case_id)`, the exact `rc` path/version/SHA-256 and diff --git a/rustfs-tier-test.sh b/rustfs-tier-test.sh index 206a712..9fd1908 100755 --- a/rustfs-tier-test.sh +++ b/rustfs-tier-test.sh @@ -239,6 +239,8 @@ suite_exit() { VALIDATION_ENV_ACTIVE=0 env_teardown "${CURRENT_TOPOLOGY}" || exit_code=1 fi + if declare -F mqtt_stop >/dev/null; then mqtt_stop || exit_code=1; fi + if declare -F mqtt_broker_stop >/dev/null; then mqtt_broker_stop || exit_code=1; fi if finish_log_capture "${exit_code}"; then exit_code=0 else @@ -289,8 +291,14 @@ NOTIFY_ENDPOINT="http://${HEAL_IP}:${RECEIVER_PORT}/hook" AUDIT_ENDPOINT="http://${HEAL_IP}:${RECEIVER_PORT}/audit" NOTIFY_TOKEN="notify-token" AUDIT_TOKEN="audit-token" -MQTT_BROKER="tcp://${HEAL_IP}:1883" +MQTT_PORT="${RUSTFS_MQTT_PORT:-1883}" +MQTT_MODE="${RUSTFS_TIER_MQTT_MODE:-auto}" +MQTT_BROKER="tcp://${HEAL_IP}:${MQTT_PORT}" MQTT_TOPIC="rustfs/events" +MQTT_SUB_PID="" +MQTT_BROKER_OWNED=0 +MQTT_DOCKER=(docker) +MQTT_CIDFILE="" RECEIVER_SCRIPT="${TEST_TMP}/rustfs-receiver.py" RECEIVER_PIDFILE="${TEST_TMP}/rustfs-receiver.pid" MQTT_PIDFILE="${TEST_TMP}/rustfs-mqtt-sub.pid" @@ -305,7 +313,7 @@ RUSTFS_USER="rustfs" RUSTFS_ADDRESS=":9000" RUSTFS_CONSOLE_ADDRESS=":9001" RUSTFS_CONSOLE_ENABLE=true -RUSTFS_OBS_LOGGER_LEVEL=error +RUSTFS_OBS_LOGGER_LEVEL=warn RUSTFS_OBS_LOG_DIRECTORY="/var/log/rustfs/" TOPO_SINGLE_SINGLE="single-single" @@ -591,6 +599,9 @@ init_artifacts() { RESULT_LOG_DIR="${ARTIFACTS_DIR}/logs" PROVENANCE_FILE="${ARTIFACTS_DIR}/provenance.json" RESIDUE_FILE="${ARTIFACTS_DIR}/residue.jsonl" + # A persistent receipt lets the caller recover an owned container after + # process cancellation or a failed Docker cleanup. + MQTT_CIDFILE="${ARTIFACTS_DIR}/mqtt-container-id" mkdir -p "${RESULTS_DIR}" "${RESULT_LOG_DIR}" if find "${RESULTS_DIR}" -maxdepth 1 -type f -name '*.json' -print -quit | grep -q .; then die "artifacts case directory is not empty; use a unique --artifacts-dir: ${RESULTS_DIR}" @@ -997,20 +1008,149 @@ receiver_stop() { sleep 1 } +mqtt_port_open() { + run_bounded 5 python3 - "${HEAL_IP}" "${MQTT_PORT}" <<'PY' +import socket +import sys + +try: + with socket.create_connection((sys.argv[1], int(sys.argv[2])), timeout=2): + pass +except OSError: + raise SystemExit(1) +PY +} + +mqtt_prepare() { + [ "${DRY_RUN}" -eq 1 ] && return 0 + case "${MQTT_MODE}" in auto|external) ;; *) die "invalid RUSTFS_TIER_MQTT_MODE: use auto or external" ;; esac + [[ "${MQTT_PORT}" =~ ^[0-9]+$ ]] && [ "${MQTT_PORT}" -ge 1 ] && [ "${MQTT_PORT}" -le 65535 ] \ + || die "RUSTFS_MQTT_PORT must be between 1 and 65535" + if ! mqtt_port_open; then + [ "${MQTT_MODE}" = auto ] || die "MQTT prerequisite unavailable at ${MQTT_BROKER} (external broker required)" + # Only manage a broker on this host. A remote endpoint must be provisioned + # by its owner; binding an ephemeral port also handles hostname resolution. + run_bounded 5 python3 - "${HEAL_IP}" <<'PY' || die "MQTT broker is unavailable and ${HEAL_IP} is not a local bind address" +import socket +import sys + +try: + with socket.socket() as sock: + sock.bind((sys.argv[1], 0)) +except OSError: + raise SystemExit(1) +PY + need_cmd docker "Docker (automatic local MQTT broker)" + if ! run_bounded 10 docker info >/dev/null 2>&1; then + MQTT_DOCKER=(sudo -n docker) + run_bounded 10 "${MQTT_DOCKER[@]}" info >/dev/null 2>&1 \ + || die "MQTT prerequisite requires a working Docker daemon or an existing broker at ${MQTT_BROKER}" + fi + [ ! -e "${MQTT_CIDFILE}" ] || die "MQTT container receipt already exists: ${MQTT_CIDFILE}" + MQTT_BROKER_OWNED=1 + run_bounded 120 "${MQTT_DOCKER[@]}" run -d --cidfile "${MQTT_CIDFILE}" \ + --name "rustfs-tier-mqtt-${RUN_TOKEN}" -p "${HEAL_IP}:${MQTT_PORT}:1883" \ + eclipse-mosquitto:2 sh -c \ + 'printf "listener 1883 0.0.0.0\nallow_anonymous true\n" > /tmp/tier-mosquitto.conf; exec mosquitto -c /tmp/tier-mosquitto.conf' \ + > "${RESULT_LOG_DIR}/mqtt-broker-start.log" 2>&1 \ + || die "MQTT broker startup failed; see ${RESULT_LOG_DIR}/mqtt-broker-start.log" + local deadline=$((SECONDS + 15)) + until mqtt_port_open; do + [ "${SECONDS}" -lt "${deadline}" ] || die "MQTT broker did not listen at ${MQTT_BROKER}" + sleep 0.2 + done + fi + mqtt_start +} + +mqtt_broker_stop() { + [ "${MQTT_BROKER_OWNED:-0}" -eq 1 ] || return 0 + local container_id + if [ -s "${MQTT_CIDFILE}" ]; then + container_id="$(cat "${MQTT_CIDFILE}")" + [[ "${container_id}" =~ ^[0-9a-f]{64}$ ]] || { warn "invalid MQTT container receipt; refusing cleanup"; return 1; } + run_bounded 10 "${MQTT_DOCKER[@]}" logs --tail 200 "${container_id}" \ + > "${RESULT_LOG_DIR}/mqtt-broker.log" 2>&1 || true + run_bounded 15 "${MQTT_DOCKER[@]}" rm -f "${container_id}" >/dev/null \ + || { warn "failed to remove owned MQTT container ${container_id}"; return 1; } + rm -f -- "${MQTT_CIDFILE}" + fi + MQTT_BROKER_OWNED=0 +} + +mqtt_check_node_access() { + [ "${DRY_RUN}" -eq 1 ] && return 0 + local node command diagnostic="${RESULT_LOG_DIR}/mqtt-network-${CURRENT_TOPOLOGY}.log" + printf -v command 'python3 -c %q %q %q' \ + 'import socket,sys; socket.create_connection((sys.argv[1], int(sys.argv[2])), timeout=3).close()' \ + "${HEAL_IP}" "${MQTT_PORT}" + : > "${diagnostic}" + for node in "$@"; do + run_bounded 10 ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" "${command}" >> "${diagnostic}" 2>&1 \ + || die "MQTT broker ${MQTT_BROKER} is unreachable from ${node}; see ${diagnostic}" + done +} + mqtt_start() { [ "${DRY_RUN}" -eq 1 ] && return 0 mqtt_stop - nohup mosquitto_sub -h "${HEAL_IP}" -p 1883 -t "${MQTT_TOPIC}" -v >> "${MQTT_EVENTS}" 2>/dev/null & - echo $! > "${MQTT_PIDFILE}" - sleep 2 - log "mqtt subscriber started (${MQTT_TOPIC})" + local probe_topic="rustfs/tier-probe/${RUN_TOKEN}" deadline=$((SECONDS + 15)) + local diagnostic="${RESULT_LOG_DIR}/mqtt-${CURRENT_TOPOLOGY:-preflight}.log" + : > "${MQTT_EVENTS}" + : > "${diagnostic}" + mosquitto_sub -h "${HEAL_IP}" -p "${MQTT_PORT}" -t "${MQTT_TOPIC}" -t "${probe_topic}" -v \ + >> "${MQTT_EVENTS}" 2>> "${diagnostic}" & + MQTT_SUB_PID=$! + echo "${MQTT_SUB_PID}" > "${MQTT_PIDFILE}" + while [ "${SECONDS}" -lt "${deadline}" ]; do + if ! kill -0 "${MQTT_SUB_PID}" 2>/dev/null; then + die "MQTT subscriber exited before readiness at ${MQTT_BROKER}; see ${diagnostic}" + fi + run_bounded 2 mosquitto_pub -h "${HEAL_IP}" -p "${MQTT_PORT}" -t "${probe_topic}" -m "${RUN_TOKEN}" \ + >> "${diagnostic}" 2>&1 || true + if grep -Fxq "${probe_topic} ${RUN_TOKEN}" "${MQTT_EVENTS}" && kill -0 "${MQTT_SUB_PID}" 2>/dev/null; then + : > "${MQTT_EVENTS}" + log "MQTT publish/subscribe readiness passed at ${MQTT_BROKER}" + return 0 + fi + sleep 0.2 + done + die "MQTT publish/subscribe readiness timed out at ${MQTT_BROKER}; see ${diagnostic}" } mqtt_stop() { - [ -f "${MQTT_PIDFILE}" ] && kill "$(cat "${MQTT_PIDFILE}")" 2>/dev/null || true - rm -f "${MQTT_PIDFILE}" - pkill -f "mosquitto_sub" 2>/dev/null || true - sleep 1 + if [ -n "${MQTT_SUB_PID:-}" ]; then + kill "${MQTT_SUB_PID}" 2>/dev/null || true + local deadline=$((SECONDS + 2)) + while kill -0 "${MQTT_SUB_PID}" 2>/dev/null && [ "${SECONDS}" -lt "${deadline}" ]; do sleep 0.1; done + kill -KILL "${MQTT_SUB_PID}" 2>/dev/null || true + wait "${MQTT_SUB_PID}" 2>/dev/null || true + MQTT_SUB_PID="" + fi + [ -z "${MQTT_PIDFILE:-}" ] || rm -f -- "${MQTT_PIDFILE}" +} + +capture_module_startup_diagnostics() { + local node line index=0 destination + for node in "${SOURCE_NODE}" "${TARGET_NODE}"; do + index=$((index + 1)) + destination="${RESULT_LOG_DIR}/module-startup-${CURRENT_TOPOLOGY}-${index}.log" + # Capture before teardown deletes the server logs. Keep only notification + # startup diagnostics, not configuration or complete service logs. + run_bounded 15 ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \ + "sudo -n sh -c 'tail -n 200 /var/log/rustfs/*.log'" 2>&1 \ + | while IFS= read -r line; do + case "${line}" in + *'notification runtime'*|*'notify runtime'*|*'mqtt target state'*|*'initialize target'*|*'initialization failed'*|*'Permission denied'*|*'No such file'*) + if printf '%s' "${line}" | grep -qiE 'password|secret|token|authorization|credential|://[^ /]+@'; then + printf '\n' + else + bounded_safe_excerpt "${line}" 1024 + printf '\n' + fi ;; + esac + done > "${destination}" || warn "module startup log collection incomplete for ${node}" + done } events_reset() { @@ -1078,6 +1218,7 @@ resolve_topology() { env_setup() { local topo="$1" url script resolve_topology "${topo}" + if [ -z "${VALIDATION_CASE}" ]; then mqtt_check_node_access "${SOURCE_NODE}" "${TARGET_NODE}"; fi url="$(resolve_package_url)" log "environment setup: topology=${topo} source=${SOURCE_NODE} target=${TARGET_NODE} endpoint=${ENDPOINT}" API_ENDPOINT="${ENDPOINT}" @@ -1152,7 +1293,11 @@ EOF [ "${VALIDATION_CASE}" = events.delivery ] && switches='{"notify_enabled":true,"audit_enabled":false}' [ "${VALIDATION_CASE}" = audit.request-correlation ] && switches='{"notify_enabled":false,"audit_enabled":true}' mbody="$(admin_api PUT /rustfs/admin/v3/module-switches "" "${switches}")"; mcode="$(admin_api_code)" - [ "${mcode}" = "200" ] || { echo "enable modules -> HTTP ${mcode}: ${mbody}"; die "failed to enable notify/audit modules"; } + if [ "${mcode}" != "200" ]; then + capture_module_startup_diagnostics + echo "enable modules -> HTTP ${mcode}: ${mbody}" + die "failed to enable notify/audit modules; startup diagnostics: ${RESULT_LOG_DIR}/module-startup-${CURRENT_TOPOLOGY}-*.log" + fi [ "${VALIDATION_CASE}" = audit.request-correlation ] || wait_targets_ready fi log "rc aliases local/target; warm bucket ${TIER_BUCKET} ready" @@ -3177,8 +3322,9 @@ main() { need_cmd python3 "Python 3 (path canonicalization and receiver)" need_cmd curl "HTTP client" need_cmd openssl "OpenSSL (SigV4)" - if [ "${VALIDATION_CASE}" != events.delivery ] && [ "${VALIDATION_CASE}" != audit.request-correlation ]; then + if [ -z "${VALIDATION_CASE}" ]; then need_cmd mosquitto_sub "mosquitto clients" + need_cmd mosquitto_pub "mosquitto clients" fi need_cmd timeout "GNU timeout (bounded CLI/network operations)" if [ "${VALIDATION_CASE}" = audit.request-correlation ] && [ "${DRY_RUN}" -eq 0 ]; then @@ -3192,6 +3338,8 @@ main() { log "nodes: ${NODES[*]} package: $(resolve_package_url)" log "rc: path=${RC_BIN} version=${RC_VERSION} sha256=${RC_SHA256:-dry-run} config=${RC_CONFIG_DIR}" + # Fail dependency preflight before purging packages, data, or server logs. + if [ -z "${VALIDATION_CASE}" ]; then mqtt_prepare; fi receiver_install local overall_fail=0 topo for topo in "${SELECTED_TOPOLOGIES[@]}"; do diff --git a/test_rustfs_tier.sh b/test_rustfs_tier.sh index 8bc6502..33fb930 100755 --- a/test_rustfs_tier.sh +++ b/test_rustfs_tier.sh @@ -1412,6 +1412,8 @@ if [ "${RUSTFS_TIER_TEST_MODE:-all}" = log-stress ]; then exit 0 fi +python3 -m unittest discover -s "${SCRIPT_DIR}" -p test_rustfs_tier_mqtt.py + test_pre_registration_before_failed_create test_cleanup_order_idempotence_and_drain test_source_bucket_cleanup_retries_bucket_not_empty_then_succeeds diff --git a/test_rustfs_tier_mqtt.py b/test_rustfs_tier_mqtt.py new file mode 100644 index 0000000..325a830 --- /dev/null +++ b/test_rustfs_tier_mqtt.py @@ -0,0 +1,245 @@ +"""Exercise the real Tier MQTT lifecycle with isolated command fixtures.""" + +import json +import os +from pathlib import Path +import shlex +import subprocess +import sys +import tempfile +import unittest + + +ROOT = Path(__file__).resolve().parent +CONTAINER_ID = "a" * 64 + + +class TierMqttTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix="tier-mqtt-test-") + self.addCleanup(self.tmp.cleanup) + self.path = Path(self.tmp.name) + self.bin = self.path / "bin" + self.bin.mkdir() + self.env = { + **os.environ, + "PATH": f"{self.bin}:{os.environ['PATH']}", + "TIER_SOURCE_ONLY": "1", + "RUSTFS_TIER_TEST_TMP": str(self.path / "tmp"), + "RUSTFS_TIER_ARTIFACTS_DIR": str(self.path / "artifacts"), + "RUSTFS_HEAL_IP": "127.0.0.1", + "RUSTFS_MQTT_PORT": "1883", + "RUSTFS_TIER_MQTT_MODE": "auto", + "MQTT_FIXTURE": str(self.path), + "MQTT_FIXTURE_MODE": "ready", + } + self.command("mosquitto_sub", """ +import json, os, pathlib, sys, time +root = pathlib.Path(os.environ['MQTT_FIXTURE']) +(root / 'subscriber-pid').write_text(str(os.getpid())) +mode = os.environ['MQTT_FIXTURE_MODE'] +if mode == 'exit': + print('Connection refused', file=sys.stderr, flush=True) + raise SystemExit(1) +while True: + probe = root / 'probe' + if mode == 'ready' and probe.exists(): + topic, payload = json.loads(probe.read_text()) + print(topic + ' ' + payload, flush=True) + probe.unlink() + time.sleep(0.01) +""") + self.command("mosquitto_pub", """ +import json, os, pathlib, sys +root = pathlib.Path(os.environ['MQTT_FIXTURE']) +args = sys.argv[1:] +(root / 'probe').write_text(json.dumps([args[args.index('-t') + 1], args[args.index('-m') + 1]])) +""") + self.command("docker", f""" +import json, os, pathlib, sys +root = pathlib.Path(os.environ['MQTT_FIXTURE']) +args = sys.argv[1:] +with (root / 'docker-calls').open('a') as out: + out.write(json.dumps(args) + '\\n') +if args[0] == 'run': + pathlib.Path(args[args.index('--cidfile') + 1]).write_text('{CONTAINER_ID}\\n') + if os.environ.get('MQTT_FIXTURE_DOCKER_FAIL'): + print('cannot bind published port', file=sys.stderr) + raise SystemExit(125) + (root / 'port-open').touch() + print('{CONTAINER_ID}') +elif args[0] == 'logs': + print('broker lifecycle diagnostic') +elif args[0] == 'rm' and os.environ.get('MQTT_FIXTURE_CLEANUP_FAIL'): + raise SystemExit(1) +""") + + def command(self, name, body): + path = self.bin / name + path.write_text(f"#!{sys.executable}\n" + body) + path.chmod(0o755) + + def run_shell(self, body, **env): + script = f""" +source {shlex.quote(str(ROOT / 'rustfs-tier-test.sh'))} +init_artifacts +{body} +""" + return subprocess.run( + ["bash", "-c", script], env={**self.env, **env}, text=True, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=25, + ) + + def docker_calls(self): + path = self.path / "docker-calls" + return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] + + def assert_subscriber_stopped(self): + pid_file = self.path / "subscriber-pid" + if pid_file.exists(): + with self.assertRaises(ProcessLookupError): + os.kill(int(pid_file.read_text()), 0) + + def test_existing_broker_requires_delivery_and_is_not_removed(self): + result = self.run_shell("mqtt_port_open() { return 0; }; mqtt_prepare") + self.assertEqual(result.returncode, 0, result.stdout) + self.assertIn("publish/subscribe readiness passed", result.stdout) + self.assertEqual(self.docker_calls(), []) + self.assertEqual((self.path / "tmp/rustfs-mqtt-events.jsonl").read_text(), "") + self.assert_subscriber_stopped() + + def test_missing_external_broker_fails_before_subscriber_start(self): + result = self.run_shell( + "mqtt_port_open() { return 1; }; mqtt_prepare", + RUSTFS_TIER_MQTT_MODE="external", + ) + self.assertNotEqual(result.returncode, 0) + self.assertIn("MQTT prerequisite unavailable", result.stdout) + self.assertFalse((self.path / "subscriber-pid").exists()) + self.assertEqual(self.docker_calls(), []) + + def test_main_rejects_missing_dependency_before_node_setup(self): + result = self.run_shell(""" +need_cmd() { :; } +resolve_rc_binary() { RC_BIN=/bin/true; } +write_provenance() { :; } +verify_rc_digest() { :; } +mqtt_port_open() { return 1; } +env_setup() { touch "$MQTT_FIXTURE/node-was-modified"; exit 99; } +main --topology single-single -y --package-url https://example.invalid/rustfs.deb +""", RUSTFS_TIER_MQTT_MODE="external") + self.assertNotEqual(result.returncode, 0) + self.assertIn("MQTT prerequisite unavailable", result.stdout) + self.assertFalse((self.path / "node-was-modified").exists()) + + def test_local_missing_broker_is_owned_and_cleaned_by_id(self): + result = self.run_shell('mqtt_port_open() { test -f "$MQTT_FIXTURE/port-open"; }; mqtt_prepare') + self.assertEqual(result.returncode, 0, result.stdout) + calls = self.docker_calls() + start = next(call for call in calls if call[0] == "run") + self.assertIn("127.0.0.1:1883:1883", start) + self.assertIn("eclipse-mosquitto:2", start) + self.assertEqual([call for call in calls if call[0] == "rm"], [["rm", "-f", CONTAINER_ID]]) + self.assertFalse((self.path / "artifacts/mqtt-container-id").exists()) + self.assertIn("broker lifecycle", (self.path / "artifacts/logs/mqtt-broker.log").read_text()) + self.assert_subscriber_stopped() + + def test_failed_container_start_retains_diagnostic_and_cleans_receipt(self): + result = self.run_shell( + "mqtt_port_open() { return 1; }; mqtt_prepare", MQTT_FIXTURE_DOCKER_FAIL="1", + ) + self.assertNotEqual(result.returncode, 0) + self.assertIn("MQTT broker startup failed", result.stdout) + self.assertIn(["rm", "-f", CONTAINER_ID], self.docker_calls()) + self.assertIn("cannot bind", (self.path / "artifacts/logs/mqtt-broker-start.log").read_text()) + + def test_term_cleans_only_owned_container_and_preserves_exit_status(self): + result = self.run_shell('mqtt_port_open() { test -f "$MQTT_FIXTURE/port-open"; }; mqtt_prepare; kill -TERM $$') + self.assertEqual(result.returncode, 143, result.stdout) + self.assertEqual([call for call in self.docker_calls() if call[0] == "rm"], [["rm", "-f", CONTAINER_ID]]) + self.assert_subscriber_stopped() + + def test_invalid_container_receipt_is_never_used_for_deletion(self): + result = self.run_shell('MQTT_BROKER_OWNED=1; printf "someone-elses-container" > "$MQTT_CIDFILE"') + self.assertNotEqual(result.returncode, 0) + self.assertIn("refusing cleanup", result.stdout) + self.assertEqual(self.docker_calls(), []) + + def test_failed_cleanup_preserves_receipt_for_a_later_recovery(self): + result = self.run_shell( + 'mqtt_port_open() { test -f "$MQTT_FIXTURE/port-open"; }; mqtt_prepare', + MQTT_FIXTURE_CLEANUP_FAIL="1", + ) + self.assertNotEqual(result.returncode, 0) + receipt = self.path / "artifacts/mqtt-container-id" + self.assertEqual(receipt.read_text().strip(), CONTAINER_ID) + recovered = self.run_shell("MQTT_BROKER_OWNED=1; mqtt_broker_stop") + self.assertEqual(recovered.returncode, 0, recovered.stdout) + self.assertFalse(receipt.exists()) + self.assertEqual([call for call in self.docker_calls() if call[0] == "rm"], [["rm", "-f", CONTAINER_ID]] * 2) + + def test_subscriber_exit_does_not_report_started(self): + result = self.run_shell("mqtt_port_open() { return 0; }; mqtt_prepare", MQTT_FIXTURE_MODE="exit") + self.assertNotEqual(result.returncode, 0) + self.assertIn("subscriber exited before readiness", result.stdout) + self.assertNotIn("readiness passed", result.stdout) + self.assertIn("Connection refused", (self.path / "artifacts/logs/mqtt-preflight.log").read_text()) + self.assert_subscriber_stopped() + + def test_live_subscriber_without_delivery_is_not_ready(self): + result = self.run_shell( + "sleep() { SECONDS=$((SECONDS + 15)); command sleep 0.02; }; mqtt_start", + MQTT_FIXTURE_MODE="quiet", + ) + self.assertNotEqual(result.returncode, 0) + self.assertIn("readiness timed out", result.stdout) + self.assertNotIn("readiness passed", result.stdout) + self.assert_subscriber_stopped() + + def test_cleanup_preserves_unrelated_subscriber(self): + other = subprocess.Popen( + [str(self.bin / "mosquitto_sub")], env={**self.env, "MQTT_FIXTURE_MODE": "quiet"}, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + try: + result = self.run_shell("mqtt_start; mqtt_stop; mqtt_stop") + self.assertEqual(result.returncode, 0, result.stdout) + self.assertIsNone(other.poll(), "cleanup terminated an unrelated subscriber") + finally: + other.terminate() + other.wait(timeout=5) + + def test_remote_missing_broker_is_not_created_locally(self): + result = self.run_shell("mqtt_port_open() { return 1; }; mqtt_prepare", RUSTFS_HEAL_IP="192.0.2.1") + self.assertNotEqual(result.returncode, 0) + self.assertIn("not a local bind address", result.stdout) + self.assertEqual(self.docker_calls(), []) + + def test_vm_network_failure_stops_before_destructive_setup(self): + self.command("ssh", "import sys; print('Connection timed out', file=sys.stderr); raise SystemExit(1)\n") + result = self.run_shell('cleanup_node() { touch "$MQTT_FIXTURE/node-was-modified"; }; env_setup single-single') + self.assertNotEqual(result.returncode, 0) + self.assertIn("unreachable from", result.stdout) + self.assertFalse((self.path / "node-was-modified").exists()) + self.assertIn("Connection timed out", next((self.path / "artifacts/logs").glob("mqtt-network-*.log")).read_text()) + + def test_module_failure_diagnostics_are_retained_and_filtered(self): + self.command("ssh", """ +print('mqtt target state target_id=primary:mqtt state=connect_timeout') +print('Local notification runtime failed: primary:mqtt: initialization failed') +print('Local notification runtime failed secret=do-not-copy-this') +print('unrelated startup configuration do-not-copy-that') +""") + result = self.run_shell("CURRENT_TOPOLOGY=single-single; capture_module_startup_diagnostics") + self.assertEqual(result.returncode, 0, result.stdout) + files = list((self.path / "artifacts/logs").glob("module-startup-*.log")) + self.assertEqual(len(files), 2) + for path in files: + text = path.read_text() + self.assertIn("state=connect_timeout", text) + self.assertIn("primary:mqtt: initialization failed", text) + self.assertNotIn("do-not-copy", text) + + +if __name__ == "__main__": + unittest.main()