ZeroMQ and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

ZeroMQ (ØMQ) — the high-performance asynchronous messaging library used in financial trading systems, AI agent pipelines, and distributed computing frameworks including the Jupyter kernel protocol — introduces three Stripe billing failure modes that do not exist in broker-based messaging systems. Unlike RabbitMQ, Kafka, or IBM MQ, ZeroMQ has no central broker, no built-in message persistence, and no native acknowledgment mechanism. Reliability is entirely the application’s responsibility. That responsibility gap creates three distinct paths to a duplicate Stripe charge, each specific to a ZeroMQ socket pattern and the metadata developers reach for when broker-provided identifiers are absent.

The first failure mode exploits the “lazy pirate” pattern — the ZeroMQ Guide’s own recommendation for reliable REQ/REP. When the REP billing service crashes after calling stripe.charges.create() but before sending the reply, the REQ client destroys and recreates its socket. Developers who capture a timestamp or socket object ID at socket-creation time and include it in the Stripe idempotency key get a different key from the new socket, and Stripe creates ch_B. The second failure mode targets PUSH/PULL pipelines: the only way to know a billing message was processed is to send an application-level ACK on a separate socket. Developers who number those ACKs and use the sequence counter in the Stripe idempotency key create a different key when the PULL worker restarts and the counter resets to zero. The third failure mode is in the ROUTER/DEALER pattern: ZeroMQ assigns a random 5-byte identity to every DEALER socket at construction time. When a DEALER reconnects, it gets a new random identity. Developers who extract that identity from the routing frame and include it in the Stripe idempotency key produce a different key on every reconnect.

This post covers all three failure modes with Python pyzmq code, content-hash idempotency keys that exclude all ZeroMQ socket metadata (socket creation timestamp, object ID, ACK counter, DEALER identity bytes), pre-flight PostgreSQL checks with ON CONFLICT DO NOTHING, per-billing-period vault keys via a spend-cap proxy capped at expected_total × 1.10, and write-before-send ordering — the two-layer governance pattern that closes all three failure modes without modifying ZeroMQ socket configuration or transport.

Failure mode 1: ZeroMQ “lazy pirate” REQ/REP socket replacement on timeout — developer captures time.time() or id(socket) at socket creation to build the Stripe idempotency key, socket is destroyed and recreated on reply timeout, new timestamp or object ID produces a different key and Stripe creates ch_B

ZeroMQ’s REQ/REP socket pair enforces strict send-receive alternation. A REQ socket after calling socket.send() must call socket.recv() before it can send again. If the REP server crashes after receiving the message but before sending a reply, the REQ socket is permanently blocked: socket.recv() will never return. Unlike TCP with a connection timeout, ZeroMQ keeps the socket open indefinitely in this state — there is no built-in request timeout in the REQ/REP pattern.

The ZeroMQ Guide’s solution is the “lazy pirate” pattern: wrap socket.recv() with a zmq.Poller timeout. If no reply arrives within the deadline, close and destroy the current REQ socket (a mere reconnect is not sufficient — the REQ socket’s internal state machine is stuck and cannot be reset), create a new socket from a new zmq.Context, and re-send the original message. This is explicitly documented as the correct recovery approach in the ZeroMQ Guide’s chapter on reliable request-reply.

The failure pattern: a billing developer builds a session discriminator when the REQ socket is created — session_ts = int(time.time() * 1000) — and includes it in the Stripe idempotency key. The reasoning sounds sound: “each socket session is a distinct context; the session timestamp namespaces retries within the same session from retries across sessions.” The result is the opposite. On the original request: REP billing service receives the message, calls stripe.charges.create() with key sha256("1722431800123:cust_123:Q3-2026"), gets back ch_A, then crashes before sending the reply. The REQ client’s poller times out. The client closes the old socket, creates a new one — new session_ts = int(time.time() * 1000) is now 1722431806500 — and re-sends the billing message. The restarted REP billing service receives it, computes sha256("1722431806500:cust_123:Q3-2026"), calls stripe.charges.create() with a different key, and Stripe creates ch_B. The customer is charged twice.

# Python pyzmq — REQ/REP lazy pirate with socket-timestamp idempotency key (unsafe)
import zmq
import time
import hashlib
import json
import stripe

def create_billing_socket(context):
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://billing-worker:5555")
    # Capture timestamp at socket construction — this changes on every socket recreate.
    session_ts = int(time.time() * 1000)
    return socket, session_ts

def send_billing_request(billing_request, retries=3):
    context = zmq.Context()
    socket, session_ts = create_billing_socket(context)
    poller = zmq.Poller()
    poller.register(socket, zmq.POLLIN)

    for attempt in range(retries):
        socket.send_json(billing_request)

        # Unsafe: idempotency key includes session_ts, which changes on socket recreate.
        # After one timeout + socket replacement, session_ts is a new value.
        idempotency_key = hashlib.sha256(
            f"{session_ts}:{billing_request['customer_id']}:{billing_request['billing_period']}"
            .encode()
        ).hexdigest()[:32]

        events = dict(poller.poll(timeout=5000))  # 5 second timeout
        if socket in events:
            reply = socket.recv_json()
            return reply
        else:
            # Timeout: lazy pirate — close and recreate the socket.
            # The old REQ socket is stuck in the send-then-wait-for-recv state;
            # zmq reconnect alone cannot reset the state machine.
            poller.unregister(socket)
            socket.setsockopt(zmq.LINGER, 0)
            socket.close()
            # Create new socket — new session_ts captured here.
            # If the REP worker called stripe.charges.create() on attempt 0 and
            # crashed before replying, this new session_ts produces a different
            # idempotency key on attempt 1, and stripe creates ch_B.
            socket, session_ts = create_billing_socket(context)
            poller.register(socket, zmq.POLLIN)

    raise Exception(f"Billing request failed after {retries} retries")


# REP side (billing worker) — crashes after stripe.charges.create() returns ch_A
# but before socket.send_json(reply) is called:
def billing_worker():
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://*:5555")

    while True:
        request = socket.recv_json()
        customer_id = request["customer_id"]
        billing_period = request["billing_period"]
        amount_cents = request["amount_cents"]
        idempotency_key = request["idempotency_key"]  # extracted from request

        # stripe.charges.create() succeeds — ch_A created.
        charge = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=request["stripe_customer_id"],
            idempotency_key=idempotency_key
        )

        # Worker crashes here (OOM kill, SIGKILL, unhandled exception).
        # The REP socket never sends the reply.
        # The REQ client's poller times out, recreates the socket,
        # resends the billing request with new session_ts → new key → ch_B.

        socket.send_json({"charge_id": charge.id, "status": "ok"})

A second variant: the developer uses id(socket) (Python’s object identity, a memory address) as the session discriminator. After the socket is destroyed and a new one created, CPython may assign a new memory address to the new socket object. id(socket) changes, the Stripe idempotency key changes, ch_B is created. A third variant: the developer uses a UUID generated at socket creation (uuid.uuid4()) as the client request ID and includes it in the idempotency key. After socket replacement, a new UUID is generated. All three variants share the same root cause: the session discriminator is captured at socket construction and changes when the lazy pirate pattern destroys and recreates the socket.

The failure is invisible in normal operation because the lazy pirate retry usually succeeds and the developer only observes one successful charge. The double-charge surfaces when (a) the REP billing worker is restarted during a billing run and (b) the billing request crossed the Stripe processing boundary but not the ZeroMQ reply boundary — a window that opens any time the REP worker crashes, is OOM-killed, or is restarted by a process supervisor.

Failure mode 2: ZeroMQ PUSH/PULL pipeline with application-level ACK counter — PULL billing worker maintains a monotonic sequence counter, counter resets to zero on worker restart, and the redelivered billing message produces a different key from ack_sequence=0 vs. the original ack_sequence=41

ZeroMQ’s PUSH/PULL socket pair distributes messages from one or more PUSH sockets to one or more PULL sockets with round-robin load balancing. There is no acknowledgment in either direction: once a PUSH socket’s socket.send() call returns, the message is in ZeroMQ’s internal send buffer and may or may not have been received and processed by the PULL side. PUSH/PULL gives you parallel delivery and load distribution; it does not give you at-least-once delivery in the sense that brokers like RabbitMQ or Kafka provide it.

Developers building billing pipelines on PUSH/PULL add their own application-level acknowledgment. A common pattern: the PULL billing worker, after successfully processing each message, sends an ACK on a separate zmq.PUSH or zmq.PAIR socket back to the producer. The producer tracks unacknowledged messages and resends after a timeout. To make ACKs and billing messages correlatable, the developer assigns a monotonic sequence number to each billing message. The producer increments a counter for each send; the consumer sends back the sequence number in the ACK. The developer also includes this sequence number in the Stripe idempotency key, reasoning: “each billing event has a unique sequence number from this producer, so the idempotency key is globally unique per billing event.”

The failure pattern: producer sends billing message with seq=41 to PULL worker. PULL worker receives it, calls stripe.charges.create() with key sha256("41:cust_123:Q3-2026"), gets back ch_A. Before the ACK is sent (or before the ACK is received by the producer), the PULL worker is killed. The producer does not receive an ACK for seq=41 within the timeout window and resends the billing message. A new PULL worker starts. The new worker’s ACK sequence counter — an in-process integer initialized to 0 at startup — is at 0. The worker does not reconstruct the sequence from the billing message or from durable state. When the worker processes the redelivered cust_123/Q3-2026 billing message, it assigns it ack_sequence=0 internally and (if using ack_sequence in the Stripe key) calls stripe.charges.create() with key sha256("0:cust_123:Q3-2026") — a different key. Stripe creates ch_B. Within 24 hours, ch_A (from sha256("41:...")) and ch_B (from sha256("0:...")) both appear in the Stripe dashboard.

# Python pyzmq — PUSH/PULL pipeline with application-level ACK counter (unsafe)
import zmq
import hashlib
import json
import stripe
import threading

# --- PULL billing worker (unsafe: ack_sequence resets to 0 on restart) ---
class BillingWorker:
    def __init__(self):
        self.context = zmq.Context()
        # Receive billing tasks from producer
        self.task_socket = self.context.socket(zmq.PULL)
        self.task_socket.connect("tcp://producer:5557")
        # Send ACKs back to producer
        self.ack_socket = self.context.socket(zmq.PUSH)
        self.ack_socket.connect("tcp://producer:5558")
        # In-process counter — resets to 0 on every worker process start.
        self.ack_sequence = 0

    def run(self):
        while True:
            task = self.task_socket.recv_json()
            customer_id = task["customer_id"]
            billing_period = task["billing_period"]
            amount_cents = task["amount_cents"]

            # Unsafe: ack_sequence is an in-process counter that resets to 0 on restart.
            # Original delivery: ack_sequence may be 41 → key = sha256("41:cust_123:Q3-2026")
            # Redelivery to new worker: ack_sequence = 0 → key = sha256("0:cust_123:Q3-2026")
            idempotency_key = hashlib.sha256(
                f"{self.ack_sequence}:{customer_id}:{billing_period}".encode()
            ).hexdigest()[:32]

            charge = stripe.Charge.create(
                amount=amount_cents,
                currency="usd",
                customer=task["stripe_customer_id"],
                idempotency_key=idempotency_key
            )

            # Worker may be killed here (OOM, SIGKILL) before ack_socket.send() is called.
            # Producer does not receive ACK, times out, resends the billing message.
            # New worker starts with ack_sequence=0, processes same message with different key → ch_B.

            self.ack_socket.send_json({
                "ack_seq": self.ack_sequence,
                "charge_id": charge.id,
                "customer_id": customer_id,
                "billing_period": billing_period
            })
            self.ack_sequence += 1


# --- PUSH producer with application-level retry on ACK timeout (unsafe) ---
class BillingProducer:
    def __init__(self):
        self.context = zmq.Context()
        self.task_socket = self.context.socket(zmq.PUSH)
        self.task_socket.bind("tcp://*:5557")
        self.ack_socket = self.context.socket(zmq.PULL)
        self.ack_socket.bind("tcp://*:5558")
        self.pending = {}  # seq → (billing_task, send_time)
        self.seq = 0

    def send_billing_task(self, billing_task):
        billing_task["seq"] = self.seq
        self.task_socket.send_json(billing_task)
        self.pending[self.seq] = (billing_task, time.time())
        self.seq += 1

    def check_acks(self):
        poller = zmq.Poller()
        poller.register(self.ack_socket, zmq.POLLIN)
        while True:
            events = dict(poller.poll(timeout=1000))
            if self.ack_socket in events:
                ack = self.ack_socket.recv_json()
                seq = ack["ack_seq"]
                if seq in self.pending:
                    del self.pending[seq]

            # Resend billing tasks whose ACK has not arrived within 30 seconds.
            now = time.time()
            for seq, (task, send_time) in list(self.pending.items()):
                if now - send_time > 30:
                    # Resending the same billing message — no new seq is assigned.
                    # A new worker receives this with ack_sequence=0 from its own counter.
                    self.task_socket.send_json(task)
                    self.pending[seq] = (task, now)

A variant uses the ZeroMQ message sequence frame directly. ZeroMQ does not attach message IDs natively (unlike AMQP’s message-id or Kafka’s offset), but developers using zmq_msg_t in C or monitoring the internal socket sequence via ZMQ_LAST_ENDPOINT sometimes derive a surrogate sequence number from the socket connection count. The same reset-on-restart failure applies. Another variant uses socket.getsockopt(zmq.LAST_ENDPOINT) to construct a unique worker identity string and includes it in the Stripe key. The endpoint string changes if the worker binds to a different port on restart, and even when the port is stable, the endpoint string alone does not make the key stable across workers — two different PULL workers bound to different ports on the same host have different endpoint strings, and both processing the same billing message (if the producer incorrectly resends to multiple workers) would produce different Stripe keys.

Failure mode 3: ZeroMQ ROUTER/DEALER pattern — DEALER socket is assigned a random 5-byte identity at construction, identity changes on DEALER reconnect, billing service extracts dealer identity from routing frame to namespace Stripe idempotency keys per client, reconnect produces ch_B

ZeroMQ’s ROUTER/DEALER socket pattern is the foundation of scalable request-reply architectures in ZMQ. A DEALER socket connects to a ROUTER socket and distributes requests across multiple workers behind the ROUTER. Unlike REQ/REP, DEALER/ROUTER does not enforce strict alternation: a DEALER can send multiple requests before receiving any replies (asynchronous). The ROUTER tracks which DEALER sent which message by prepending the DEALER’s identity to every received message as a routing frame.

ZeroMQ assigns each DEALER socket a random identity at construction time. By default, the identity is a 5-byte value beginning with a zero byte followed by 4 random bytes: b"\x00\xf3\xa2\x11\x8c". The developer can set a static identity with socket.setsockopt(zmq.IDENTITY, b"my-stable-id") before connecting, but this is not the default behavior and is frequently omitted. When a DEALER disconnects and reconnects (network interruption, process restart, container recreation), the new DEALER socket gets a new random identity. The ROUTER billing service, which receives messages as multipart envelopes [dealer_identity, b"", json_payload], sees a different dealer_identity for the reconnected client.

The failure pattern: a developer builds a billing client using a DEALER socket. The developer extracts the DEALER’s own identity (socket.getsockopt(zmq.IDENTITY)) and includes it in the Stripe idempotency key, reasoning: “the identity uniquely identifies this client process; two clients with the same customer and billing period are different billing events.” On the original request: DEALER sends [b"", json_payload] (DEALER prepends an empty delimiter). ROUTER receives [dealer_identity_1, b"", json_payload]. Billing worker calls stripe.charges.create() with key sha256(dealer_identity_1.hex() + ":cust_123:Q3-2026") and gets back ch_A. The billing worker crashes before sending the reply. The DEALER’s application-level retry fires. The DEALER reconnects (new random identity: dealer_identity_2) and re-sends the billing message. The ROUTER receives it with dealer_identity_2 in the routing frame. The billing worker builds sha256(dealer_identity_2.hex() + ":cust_123:Q3-2026") — a completely different key — and stripe.charges.create() creates ch_B.

# Python pyzmq — ROUTER/DEALER pattern with DEALER identity in Stripe key (unsafe)
import zmq
import hashlib
import json
import stripe

# --- DEALER client (billing request sender) ---
class BillingClient:
    def __init__(self):
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.DEALER)
        # No static identity set — random 5-byte identity assigned by ZMQ.
        # If set here (before connect), it persists across reconnects on the SAME socket.
        # But if the DEALER socket is destroyed and recreated (connection loss handling),
        # a new random identity is assigned.
        self.socket.connect("tcp://billing-router:5559")
        # Capture own identity at construction time.
        # Changes if the socket is recreated (network failure, crash+restart, container restart).
        self.client_id = self.socket.getsockopt(zmq.IDENTITY).hex()

    def send_billing_request(self, customer_id, billing_period, amount_cents, stripe_customer_id):
        # Unsafe: self.client_id changes when this DEALER socket is destroyed and recreated.
        # A reconnected DEALER has a new random identity → different Stripe key → ch_B.
        idempotency_key = hashlib.sha256(
            f"{self.client_id}:{customer_id}:{billing_period}".encode()
        ).hexdigest()[:32]

        payload = {
            "customer_id": customer_id,
            "billing_period": billing_period,
            "amount_cents": amount_cents,
            "stripe_customer_id": stripe_customer_id,
            "idempotency_key": idempotency_key  # derived from client_id
        }
        # DEALER sends: [b"", json_payload] — ZMQ adds routing frame automatically
        self.socket.send_multipart([b"", json.dumps(payload).encode()])


# --- ROUTER billing service (billing worker behind router) ---
class BillingRouter:
    def __init__(self):
        self.context = zmq.Context()
        self.router = self.context.socket(zmq.ROUTER)
        self.router.bind("tcp://*:5559")

    def run(self):
        while True:
            # ROUTER receives: [dealer_identity, b"", json_payload]
            frames = self.router.recv_multipart()
            dealer_identity = frames[0]   # random 5-byte value — changes on DEALER reconnect
            empty = frames[1]             # delimiter frame
            payload = json.loads(frames[2])

            # Some developers use dealer_identity at the ROUTER side instead of (or in addition to)
            # the idempotency_key sent by the client:
            # billing_key = sha256(dealer_identity.hex() + ":" + payload["customer_id"] + ...)
            # If the DEALER reconnects and retries, dealer_identity is a new random value.

            idempotency_key = payload["idempotency_key"]  # from client, already derived from client_id

            charge = stripe.Charge.create(
                amount=payload["amount_cents"],
                currency="usd",
                customer=payload["stripe_customer_id"],
                idempotency_key=idempotency_key
            )

            # If the worker crashes here, dealer_identity on the retry is different.
            # The client resends with new self.client_id → new idempotency_key → ch_B.

            reply = json.dumps({"charge_id": charge.id, "status": "ok"}).encode()
            self.router.send_multipart([dealer_identity, b"", reply])


# What identity change looks like at the socket level:
#
# First DEALER connect (socket 1):
#   DEALER identity = b"\x00\xf3\xa2\x11\x8c"  → hex = "00f3a2118c"
#   idempotency_key = sha256("00f3a2118c:cust_123:Q3-2026") → "a8f3bc..."
#   stripe.charges.create() → ch_A (success, worker crashes before reply)
#
# DEALER reconnects (socket destroyed + recreated, new random identity):
#   DEALER identity = b"\x00\x7b\x5d\x29\xe1"  → hex = "007b5d29e1"
#   idempotency_key = sha256("007b5d29e1:cust_123:Q3-2026") → "f94c21..."  ← different
#   stripe.charges.create() → ch_B (duplicate charge)
#
# Fix: set a static identity before connect:
#   socket.setsockopt(zmq.IDENTITY, b"billing-client-prod-1")
#   ... and exclude it from the Stripe key even then (it still changes on container restart
#   if the identity is derived from a pod name, hostname hash, or process ID).

A related variant: the developer does not use socket.getsockopt(zmq.IDENTITY) from the DEALER side, but instead extracts the routing identity from the ROUTER side’s received frames (frames[0]) and uses it to namespace the Stripe key in a stateful billing service. The ROUTER sees a different frames[0] for the reconnected DEALER’s retry message, builds a different Stripe key, and creates ch_B. Whether the identity is accessed from the DEALER or extracted from the ROUTER’s routing envelope, the root cause is the same: the ZMQ routing identity is a per-socket-construction value, not a per-application or per-client-process value. It changes whenever the DEALER socket is destroyed and reconstructed.

Setting a static identity via setsockopt(zmq.IDENTITY, ...) before connect() prevents the ZMQ layer from assigning a random identity, but does not solve the idempotency problem if the static identity is itself derived from an unstable value: container hostname (changes on pod restart), process ID (changes on restart), or a UUID generated at startup without durable persistence. Only a truly static, durable identity value — hardcoded or loaded from persistent config — would make a ZMQ-identity-based Stripe key stable across restarts. Using a content-hash key that does not include the ZMQ identity is simpler and more robust.

What doesn’t prevent duplicate Stripe charges in ZeroMQ

The fix: content-hash idempotency key + pre-flight database check + vault key spend cap

Three changes close all three failure modes without modifying ZeroMQ socket patterns, transport layer, or network topology.

1. Content-hash idempotency key derived exclusively from stable business fields

The Stripe idempotency key must be derived from fields that are stable across every ZeroMQ delivery path: the initial delivery, lazy pirate retry (new socket, new timestamp), ACK timeout redelivery (new worker, counter reset to zero), and DEALER reconnect (new random identity). The billing message payload’s business fields — customer ID, billing period, and a service-level namespace — meet all conditions. ZeroMQ socket metadata — socket creation timestamps, object IDs, ACK sequence counters, DEALER identity bytes, ZMQ sequence numbers — does not.

# Python pyzmq — safe content-hash idempotency key, all three socket patterns
import zmq
import hashlib
import json
import psycopg2
import stripe

def content_hash_key(customer_id: str, billing_period: str) -> str:
    # Inputs: stable business fields only.
    # Excluded (all ZMQ socket metadata):
    #   - time.time() captured at socket construction (changes on lazy pirate recreate)
    #   - id(socket) / object identity (changes on new socket object)
    #   - ack_sequence (in-process counter, resets to 0 on worker restart)
    #   - socket.getsockopt(zmq.IDENTITY) (random per-socket-construction, changes on DEALER reconnect)
    #   - routing frame identity from recv_multipart()[0] (same as above)
    #   - ZMQ last endpoint string (changes if port changes on restart)
    raw = f"{customer_id}:{billing_period}:zmq-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]


# --- Safe REP billing worker (lazy pirate compatible) ---
class SafeBillingWorker:
    def __init__(self, pg_conn):
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.REP)
        self.socket.bind("tcp://*:5555")
        self.pg_conn = pg_conn  # external PostgreSQL — survives worker restarts

    def run(self):
        while True:
            request = self.socket.recv_json()
            customer_id = request["customer_id"]
            billing_period = request["billing_period"]
            amount_cents = request["amount_cents"]
            stripe_customer_id = request["stripe_customer_id"]

            # Safe idempotency key: excludes ALL ZMQ socket metadata.
            idempotency_key = content_hash_key(customer_id, billing_period)

            # Pre-flight check: INSERT into external PostgreSQL before calling Stripe.
            # ON CONFLICT DO NOTHING: if this (customer_id, billing_period) pair already
            # exists (because a previous delivery called stripe.charges.create() before
            # the worker crashed), rows_affected = 0 and we skip Stripe.
            # This closes FM2 when the ACK timeout window exceeds Stripe's 24-hour
            # idempotency cache, and FM1/FM3 where Stripe returns ch_A if called within
            # 24 hours of the original charge (the content-hash key matches).
            with self.pg_conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO billing_records (customer_id, billing_period, idempotency_key) "
                    "VALUES (%s, %s, %s) ON CONFLICT (customer_id, billing_period) DO NOTHING",
                    (customer_id, billing_period, idempotency_key)
                )
                rows_affected = cur.rowcount
                self.pg_conn.commit()

            if rows_affected == 0:
                # Pre-flight row already exists: a prior delivery called stripe.charges.create().
                # Skip Stripe and send a success reply — the message was already processed.
                self.socket.send_json({"status": "already_processed"})
                continue

            # Write-before-call: billing_records row is committed BEFORE stripe.charges.create().
            # If worker is killed between PostgreSQL commit and Stripe call:
            #   - Stripe was never called (customer is not charged — correct).
            #   - Pre-flight row exists on next delivery → Stripe call is safely skipped.
            # If worker is killed between Stripe call and socket.send_json(reply):
            #   - ch_A exists in Stripe.
            #   - Pre-flight row exists → next delivery skips Stripe → no ch_B.
            charge = stripe.Charge.create(
                amount=amount_cents,
                currency="usd",
                customer=stripe_customer_id,
                idempotency_key=idempotency_key
            )

            with self.pg_conn.cursor() as cur:
                cur.execute(
                    "UPDATE billing_records SET charge_id = %s WHERE customer_id = %s AND billing_period = %s",
                    (charge.id, customer_id, billing_period)
                )
                self.pg_conn.commit()

            # If this send fails (worker killed here), the lazy pirate client retries.
            # The retry lands on a new worker with ack_sequence=0 or a new DEALER identity,
            # but the Stripe key is the SAME content-hash, and the pre-flight row exists
            # (rowsAffected=0) → Stripe call is skipped → no ch_B.
            self.socket.send_json({"charge_id": charge.id, "status": "ok"})

2. Pre-flight database check with ON CONFLICT DO NOTHING

The content-hash idempotency key closes failure modes 1 and 3 when the retry happens within Stripe’s 24-hour idempotency window: Stripe returns ch_A on every call with the same key within 24 hours, regardless of the ZMQ socket pattern that triggered the call. Failure mode 2 (ACK timeout redelivery) can occur outside the 24-hour window when billing message queues back up or PULL workers are offline for extended periods — common in batch billing pipelines that run overnight.

The pre-flight PostgreSQL check closes all three failure modes regardless of the 24-hour window. Before calling stripe.charges.create(), the billing worker executes INSERT INTO billing_records (customer_id, billing_period) VALUES (%s, %s) ON CONFLICT (customer_id, billing_period) DO NOTHING and inspects cur.rowcount. If rowcount == 1 (row inserted), the worker calls Stripe and commits. If rowcount == 0 (row already existed), the worker skips Stripe and commits the ZMQ message. The PostgreSQL table is external to the ZeroMQ process — it survives lazy pirate socket replacements, PULL worker restarts, and DEALER reconnects. The unique constraint on (customer_id, billing_period) is the durable source of truth that outlasts any ZMQ socket session.

3. Per-billing-period vault key capped at expected_total × 1.10

A Keybrake vault key issued per billing period with a spend cap of expected_total × 1.10 provides a hard backstop for the case where the pre-flight PostgreSQL check is temporarily unavailable during a lazy pirate retry or ACK timeout redelivery. If PostgreSQL is transiently unreachable (network partition between the billing worker and the database, rolling maintenance, connection pool exhaustion), the pre-flight check cannot execute. Without the vault cap, stripe.charges.create() might proceed with a different ZMQ-metadata-derived key and create ch_B. With the vault cap, the Keybrake proxy compares the attempted charge against the period total and rejects any amount that would push the period beyond the 110% cap. The vault cap acts at the Stripe API boundary, downstream of both the ZMQ transport and the pre-flight check, providing defense in depth when both upstream layers fail simultaneously.

Gap analysis: ZeroMQ-specific scenarios not covered by simpler approaches

ZeroMQ zmq_proxy() forwarder restart drops in-transit messages, forcing application retry with a timer-derived idempotency key

Many ZeroMQ billing architectures use a zmq_proxy() (or the Python zmq.device(zmq.QUEUE, frontend, backend) equivalent) as a message forwarder between producers and workers. The proxy itself is a stateless in-process forwarder: it reads from the frontend socket and writes to the backend socket. Messages in the proxy’s internal buffers at the moment the proxy process is killed are lost — they have been dequeued from the producer’s send buffer but have not yet been acknowledged by the worker. The producer’s application-level retry timer fires (because no ACK arrived), and the producer resends the billing message. If the producer’s retry includes a new timer-capture timestamp in the Stripe key (because the original timer-based key is now “stale” from the producer’s perspective), the retry has a different key. The content-hash key and pre-flight check close this case regardless of how many times the zmq_proxy() is restarted between the original send and the final successful processing.

ZeroMQ PUB/SUB billing notification with subscriber reconnect and publisher-side replay using sequence numbers

Some billing notification architectures use ZeroMQ PUB/SUB: a billing event publisher sends notifications to subscriber billing workers. PUB/SUB in ZeroMQ has no persistence — messages published before a subscriber connects are lost. When a subscriber reconnects after a disconnect, it misses any messages published during the disconnect. Developers add a separate replay mechanism: the publisher keeps a bounded in-memory log of the last N messages and serves replays from a REP socket. The subscriber, on reconnect, queries the replay socket for messages since the last sequence number it received.

The failure: the replay mechanism assigns a new publisher-side sequence number to replayed messages, different from the original publication sequence. If the subscriber uses the sequence number from the SUB socket (the original) in the Stripe key but receives replays from the REP socket with different sequence numbers, the two delivery paths produce different keys. The content-hash key (which does not reference any ZMQ sequence or replay sequence) is stable across both delivery paths. The pre-flight check provides additional protection when the replay occurs outside the 24-hour Stripe idempotency window.

ZeroMQ PUSH socket with zmq.NOBLOCK flag raises zmq.Again at HWM, application retries the send, timing-based key differs between first send attempt and retry

When a PUSH socket’s send high water mark (ZMQ_SNDHWM) is reached, calling socket.send(zmq.NOBLOCK) raises zmq.Again (errno EAGAIN) instead of blocking. Some billing producer implementations catch zmq.Again, wait a short interval, and retry the send. If the retry logic generates a new timer-based request ID for the Stripe idempotency key on each send() attempt, the key differs between the first failed send (which raised EAGAIN) and the successful retry. This is a producer-side failure mode, not a consumer-side one: the message has not yet been delivered to any worker, and no Stripe call has been made, so there is no duplicate charge yet. The risk is that the application logs two “billing request sent” events with different request IDs, creating confusion in the audit trail. The content-hash key (generated from the billing event’s payload fields, not from the send attempt metadata) is identical across all retries of the same socket.send() call.

Frequently asked questions

Can I prevent the lazy pirate key change by including the idempotency key in the ZMQ message payload itself and computing it before the first send?

Yes — and this is the correct approach. If the billing producer computes the Stripe idempotency key from stable business fields (sha256(customer_id + ":" + billing_period + ":zmq-billing")[:32]) before the first socket.send() call and includes it in the message payload JSON, then every lazy pirate retry sends the same idempotency key. The billing worker extracts the key from the payload rather than computing it from socket metadata. The key is identical on every retry, and Stripe returns ch_A on all retries within 24 hours. This is what the content-hash key pattern achieves when applied at the producer: the key is derived from the event content, not from the ZMQ transport layer, so it survives socket recreation.

Does setting socket.setsockopt(zmq.IDENTITY, b"stable-billing-client-1") on the DEALER before connect() fully prevent failure mode 3?

It prevents FM3 if and only if the same static identity bytes are configured for every instance of the DEALER process across all deployments and restarts. If the identity is derived from a container hostname (os.environ.get("HOSTNAME").encode()), a pod name ("billing-pod-abc123".encode()), a process ID (str(os.getpid()).encode()), or a UUID generated at startup (str(uuid.uuid4()).encode()), the identity changes on pod restart, container recreation, or process restart — precisely the events that trigger the billing retry. A truly static identity (hardcoded or loaded from a secret that does not change between deployments) would make the ZMQ routing identity stable. In practice, keeping a static identity consistent across an entire fleet of billing workers is operationally fragile: configuration drift, new deployments with different naming conventions, or blue-green deploys introduce identity collisions (two different workers with the same identity) or identity changes (the old worker’s identity is not the new worker’s). A content-hash Stripe key that excludes the ZMQ identity entirely is simpler and does not depend on any ZMQ configuration.

Does the ZeroMQ “majordomo” or “paranoid pirate” pattern add enough reliability to prevent duplicate Stripe charges?

The Paranoid Pirate pattern (DEALER/ROUTER with heartbeats between workers and a broker) adds worker presence detection and task redistribution. When a worker fails mid-task, the broker detects the missing heartbeat and redistributes the task to another available worker. This is an improvement over the simple Lazy Pirate, but it does not prevent the duplicate Stripe charge: the original worker may have called stripe.charges.create() successfully before the heartbeat stopped. The redistributed task arrives at a new worker, which calls stripe.charges.create() again. The content-hash key is the same on both calls (if the new worker uses content-hash keys), and Stripe returns ch_A on the second call within 24 hours — but if the redistribution happens more than 24 hours after the original (unusual but possible during extended outages), Stripe’s idempotency cache has expired and creates ch_B. The pre-flight PostgreSQL check closes the post-24-hour case regardless of which ZMQ reliability pattern (Lazy Pirate, Paranoid Pirate, Majordomo) is used.

Can I use ZeroMQ’s zmq.PAIR socket for a bidirectional billing channel and include the PAIR socket endpoint in the Stripe idempotency key?

zmq.PAIR sockets are point-to-point and are often used for thread-to-thread or process-to-process communication where exactly one producer and one consumer are always present. The PAIR socket has no routing frame and no built-in identity. If the developer uses the PAIR socket’s endpoint address (socket.getsockopt_string(zmq.LAST_ENDPOINT)) as a component of the Stripe idempotency key, the key is stable as long as the endpoint address does not change. On a single-process deployment where the PAIR socket always binds to the same port (tcp://127.0.0.1:5560), the endpoint is stable. On container-based deployments where the port is dynamically assigned or the hostname changes on restart, the endpoint changes. In all cases, using the endpoint in the Stripe key is unnecessary risk: the content-hash key derived from billing payload fields is stable regardless of endpoint, port, or socket type.

If the billing worker calls stripe.charges.create() and the call times out (stripe.error.Timeout), should it retry the Stripe call or skip it?

A stripe.error.Timeout means the Stripe API server may or may not have processed the charge: the request was sent but the response was not received before the SDK’s timeout. If the billing worker retries the Stripe call immediately with the same content-hash idempotency key, and Stripe processed the original request, Stripe returns ch_A — no duplicate charge. If Stripe did not process the original request (the server received the TCP connection but timed out before generating a response), Stripe processes the retry and creates ch_A — correct behavior. The content-hash idempotency key makes it safe to retry immediately on stripe.error.Timeout: Stripe’s idempotency cache handles the “which case am I in” question. The billing worker should not skip the Stripe call on timeout (the customer might go uncharged) and should not change the idempotency key on timeout retry (that would create ch_B). Within the ZeroMQ context, if the stripe.error.Timeout causes the REP worker to fail to send a reply (raising an exception that the worker’s error handler catches and proceeds to ZMQ reply without retrying Stripe), the lazy pirate client will retry the entire request on the ZMQ layer. The pre-flight PostgreSQL check on the second delivery handles the case where the first Stripe call succeeded despite the timeout.