Publish from a relay

The relay is a small loop with three steps. Read the pending rows, publish each message, and mark the row sent after the broker acknowledges it. Marking after acknowledgement is what makes a crash in the middle harmless.

def relay_tick(conn, broker):
    for row in conn.execute("select id, topic, body from outbox where sent is null"):
        if broker.publish(row["body"]):
            conn.execute("update outbox set sent = 1 where id = ?", (row["id"],))

The relay can publish twice when it crashes between sending and marking. That is expected. The guarantee is at least once delivery, so the consumer deduplicates on a stable message id.

Keep the message id stable across retries. A new id on each attempt makes the duplicate unrecognisable.

Alert on the pending count. A backlog that keeps growing is the earliest sign that publishing has stopped.