RabbitMQ quorum queues: confirms, redelivery and idempotent effects

Provisioned staging client with Pika1.3+

import getpass, os, sqlite3, ssl, uuid, time
from pathlib import Path
import pika

root = Path('/home/academy/rabbitmq-probe-20261028')
ledger = root / 'effect-ledger.sqlite'
fd = os.open(ledger, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
os.close(fd)
db = sqlite3.connect(ledger)
db.execute('PRAGMA synchronous=FULL')
db.execute('CREATE TABLE effect(message_id TEXT PRIMARY KEY, body BLOB NOT NULL)')
db.commit()
password = getpass.getpass('Staging RabbitMQ password: ')
tls = ssl.create_default_context(cafile='/etc/ssl/certs/academy-rabbitmq-ca.pem')
def connect():
    return pika.BlockingConnection(pika.ConnectionParameters(
        host='rabbitmq.example.com', port=5671, virtual_host='academy_lab',
        credentials=pika.PlainCredentials('academy_probe', password),
        ssl_options=pika.SSLOptions(tls, 'rabbitmq.example.com'),
        heartbeat=30, blocked_connection_timeout=10, socket_timeout=5,
        connection_attempts=1))

queue = 'academy_ack_probe_20261028'
message_id = 'academy_effect_20261028'
conn = connect()
ch = conn.channel()
try:
    ch.queue_declare(queue=queue, passive=True)
except pika.exceptions.ChannelClosedByBroker as exc:
    if exc.reply_code != 404:
        raise
    ch = conn.channel()
else:
    raise RuntimeError('Queue already exists: stop; do not reuse it')
ch.queue_declare(queue=queue, durable=True,
                 arguments={'x-queue-type':'quorum', 'x-quorum-initial-group-size':3})
ch.confirm_delivery()
props = pika.BasicProperties(delivery_mode=2, message_id=message_id,
                             content_type='text/plain')
try:
    ch.basic_publish(exchange='', routing_key='absent_' + uuid.uuid4().hex,
                     body=b'negative-route-probe', properties=props, mandatory=True)
except pika.exceptions.UnroutableError:
    print('PASS: unroutable mandatory publish reported')
else:
    raise RuntimeError('Missing negative routing result')
ch.basic_publish(exchange='', routing_key=queue, body=b'academy-synthetic-effect',
                 properties=props, mandatory=True)
print('PASS: routed publish confirmed')
conn.close()

for attempt in (1,2):
    conn = connect()
    ch = conn.channel()
    deadline = time.monotonic() + 10
    while True:
        method, properties, body = ch.basic_get(queue=queue, auto_ack=False)
        if method is not None or time.monotonic() >= deadline:
            break
        conn.process_data_events(time_limit=0.1)
        time.sleep(0.1)
    if method is None or properties.message_id != message_id or body != b'academy-synthetic-effect':
        raise RuntimeError('Unexpected delivery: preserve state and inspect')
    if attempt == 2 and not method.redelivered:
        raise RuntimeError('Expected redelivery after unacknowledged close')
    with db:
        db.execute('INSERT OR IGNORE INTO effect VALUES (?,?)',(message_id,body))
        stored = db.execute('SELECT body FROM effect WHERE message_id=?',(message_id,)).fetchone()
        if stored[0] != body:
            raise RuntimeError('Message-ID collision with different effect')
    if attempt == 1:
        print('PASS: synthetic effect committed; closing without ACK')
    else:
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print('PASS: redelivery deduplicated, then ACK sent')
    conn.close()
assert db.execute('SELECT count(*) FROM effect').fetchone()[0] == 1
db.close()
print('PASS: exactly one synthetic ledger row; not an exactly-once service claim')