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

NSQ — the real-time distributed messaging platform built by Bitly and used across ad-tech, fintech, and AI agent pipelines — introduces three Stripe billing failure modes that differ from those found in broker-centric systems like RabbitMQ or Kafka. NSQ’s architecture is deliberately decentralized: each nsqd daemon stores messages independently, consumers discover nsqd instances via nsqlookupd at startup and periodically thereafter, and message routing is handled at the consumer layer. That decentralized model shifts delivery-tracking responsibility onto the application and creates three distinct paths to a duplicate Stripe charge, each rooted in metadata that developers reach for to uniquely identify billing events when an NSQ-native identifier is absent or unstable.

The first failure mode is the most common. NSQ attaches a per-message Attempts counter (a uint16 in the NSQ wire protocol) to every message. It starts at 1 on first delivery and increments by 1 on every requeue and redelivery — including the requeue triggered when a consumer crashes after stripe.charges.create() returns ch_A but before message.Finish() is called. Developers who include msg.Attempts in the Stripe idempotency key to “distinguish retries from original deliveries” get a different key on every redelivery, and Stripe creates a new charge for each one. With NSQ’s default retry behavior and a consumer that keeps crashing at the same point, a single billing message can produce five or more distinct Stripe charges in a single billing run. The second failure mode targets consumers that scope a UUID to an nsqd connection: when the connection drops and the consumer reconnects, the UUID changes, and a billing message redelivered from nsqd’s disk persistence is processed with a different key and creates ch_B. The third failure mode is specific to pynsq: stripe.Charge.create() called synchronously in a Tornado IOLoop handler blocks the event loop, preventing TOUCH keepalives from being sent, causing nsqd to requeue the message while Stripe is still processing the first charge.

This post covers all three failure modes with Go (go-nsq) and Python (pynsq) code, content-hash idempotency keys that exclude all NSQ delivery metadata (Attempts counter, connection UUID, nsqd address, per-delivery timestamp), 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 the IOLoop-safe async pattern for pynsq billing handlers — the three-layer governance pattern that closes all three failure modes without modifying NSQ topic/channel configuration, nsqd deployment topology, or consumer concurrency settings.

Failure mode 1: NSQ Message.Attempts counter in Stripe idempotency key — counter starts at 1 on first delivery and increments on every requeue including crash-recovery requeues after a successful Stripe call, producing ch_B through ch_N with each retry

Every NSQ message carries an Attempts field exposed to consumers via the go-nsq nsq.Message struct (m.Attempts, a uint16) and the pynsq nsq.Message object (message.attempts). On the first delivery of a message from nsqd to a consumer, Attempts is 1. Every time nsqd requeues a message — because the consumer sent a REQ command, because the consumer’s msg-timeout expired without a FIN or REQ, or because the consumer connection dropped while the message was in-flight — nsqd increments Attempts by 1 before the next delivery. A message requeued four times arrives on its fifth delivery with Attempts=5.

The billing idempotency anti-pattern: developers want to distinguish “this is the original billing attempt” from “this is a retry after a failure” and include msg.Attempts in the Stripe idempotency key. The reasoning sounds defensive — “if the original call somehow went through without my knowing, I want a different key on the retry so I don’t accidentally block a legitimate new charge.” The result is the opposite: every redelivery bypasses Stripe’s 24-hour idempotency cache and creates a fresh charge. On first delivery: consumer receives the billing message with Attempts=1, computes sha256("cust_123:Q3-2026:1") as the idempotency key, calls stripe.charges.create(), gets back ch_A, crashes before calling message.Finish(). nsqd’s msg-timeout (60 seconds by default) expires. nsqd requeues the message with Attempts=2. A consumer receives it, computes sha256("cust_123:Q3-2026:2") — a completely different hash — calls stripe.charges.create() with this new key, and Stripe creates ch_B. The billing message is now in-flight again. The consumer crashes again at the same point (say, a database write failure after Stripe succeeds). Attempts=3. ch_C. The customer is charged three times in sixty-second intervals.

// Go go-nsq — HandleMessage with Attempts in idempotency key (unsafe)
package main

import (
    "crypto/sha256"
    "encoding/json"
    "fmt"
    "strconv"

    "github.com/nsqio/go-nsq"
    "github.com/stripe/stripe-go/v76"
    "github.com/stripe/stripe-go/v76/charge"
)

type BillingEvent struct {
    CustomerID       string `json:"customer_id"`
    BillingPeriod    string `json:"billing_period"`
    AmountCents      int64  `json:"amount_cents"`
    StripeCustomerID string  `json:"stripe_customer_id"`
}

type BillingHandler struct{}

func (h *BillingHandler) HandleMessage(m *nsq.Message) error {
    var event BillingEvent
    if err := json.Unmarshal(m.Body, &event); err != nil {
        return err
    }

    // Unsafe: m.Attempts starts at 1 on first delivery.
    // Increments on EVERY requeue — including the requeue triggered by a
    // consumer crash after stripe.Charges.New() returns ch_A but before
    // m.Finish() is called. Each attempt produces a different Stripe key.
    //
    // Delivery 1: Attempts=1 → sha256("cust_123:Q3-2026:1") → ch_A (crash before Finish)
    // Delivery 2: Attempts=2 → sha256("cust_123:Q3-2026:2") → ch_B (crash again)
    // Delivery 3: Attempts=3 → sha256("cust_123:Q3-2026:3") → ch_C
    idempotencyKey := fmt.Sprintf("%x", sha256.Sum256([]byte(
        event.CustomerID+":"+event.BillingPeriod+":"+strconv.Itoa(int(m.Attempts)),
    )))[:32]

    params := &stripe.ChargeParams{
        Amount:   stripe.Int64(event.AmountCents),
        Currency: stripe.String(string(stripe.CurrencyUSD)),
        Customer: stripe.String(event.StripeCustomerID),
    }
    params.IdempotencyKey = stripe.String(idempotencyKey)

    _, err := charge.New(params)
    if err != nil {
        // Returning err causes go-nsq to requeue the message with m.Attempts+1.
        // If stripe.Charges.New() already created ch_A before returning this error
        // (e.g., stripe.ErrAPIConnection after the server processed the request),
        // the next delivery creates ch_B.
        return err
    }

    // If this handler panics or the process is killed here (after charge.New succeeds
    // but before m.Finish()), go-nsq/nsqd requeues with m.Attempts+1 on msg-timeout.
    m.Finish()
    return nil
}

// nsqd msg-timeout default: 60 seconds.
// go-nsq default MaxAttempts: 0 (unlimited retries).
// A billing consumer that crashes at the Stripe→Finish boundary on every attempt
// will create ch_A, ch_B, ch_C, ... with 60-second intervals between charges.

The failure is invisible in monitoring dashboards that only track “billing messages processed.” The go-nsq handler returns nil zero times (it crashes before return nil), so the consumer’s FinishCount stat stays at zero and the billing message keeps cycling. Operations teams see the message being consumed and requeued in an nsqadmin loop without realizing that each iteration created a new Stripe charge. The pattern is especially dangerous in batch billing runs where hundreds of customers are processed simultaneously: a single OOM kill at the Stripe call site can cause the entire batch to cycle, with each customer’s billing message requeued and a second charge created for each.

A variant: developers use m.Timestamp (nanoseconds since epoch, set at message publication time) or m.ID (the 16-byte unique message ID assigned by nsqd at publication) as components of the Stripe idempotency key, combined with a processing-time timestamp from time.Now(). m.Timestamp and m.ID are stable across redeliveries — they are set at publish time and do not change on requeue. But if the developer generates a per-handler-invocation timestamp (time.Now().UnixNano() inside HandleMessage) and combines it with the message ID, the combination is different on every delivery. The message ID alone would be a safe key component; the combination with a per-invocation timestamp defeats it.

Failure mode 2: NSQ consumer connection UUID scoped to an nsqd connection — UUID changes when the consumer reconnects after a crash or network partition, billing message redelivered from nsqd disk persistence is processed with a different UUID and Stripe creates ch_B

NSQ consumers discover nsqd instances via nsqlookupd and establish a separate TCP connection to each nsqd daemon that has messages for the subscribed topic. In go-nsq, each connection is represented by a *nsq.Conn object. In pynsq, the Reader class maintains a connection per nsqd address as a Tornado IOStream. Developers who want to “scope” Stripe idempotency keys to a specific consumer session — reasoning that two different consumer processes handling the same logical billing event are different billing actions — sometimes generate a UUID at connection establishment time and include it in the idempotency key.

The connection UUID survives for as long as the connection to that specific nsqd daemon persists. When the nsqd daemon restarts (OOM kill, rolling deploy, network interruption causing TCP timeout), the consumer’s connection is dropped. The consumer’s go-nsq ConnectToNSQD or pynsq Reader.connect_to_nsqd() re-establishes the connection. At reconnect, a developer who generates the UUID in a connection callback (go-nsq’s OnConnect delegate or pynsq’s connect_callback) generates a new UUID. The old UUID is gone — it was never persisted, only held in memory for the lifetime of the previous connection object.

The failure pattern: consumer establishes connection to nsqd-A, generates conn_uuid = "a1b2c3...", receives billing message for cust_123/Q3-2026 from nsqd-A. Consumer computes sha256("a1b2c3...:cust_123:Q3-2026") as the Stripe idempotency key, calls stripe.charges.create(), gets back ch_A. nsqd-A crashes immediately after (or OOM-kills). The consumer’s connection to nsqd-A is dropped before the consumer can send FIN. nsqd-A restarts from its disk persistence (--data-path): the billing message was stored to disk but was never FINished, so nsqd-A restores it to the in-flight queue and makes it available for delivery again. The consumer reconnects to nsqd-A and receives the billing message again — this time generating conn_uuid = "f9e8d7..." (a new UUID from the new connection callback). The idempotency key is now sha256("f9e8d7...:cust_123:Q3-2026") — a completely different hash. stripe.charges.create() creates ch_B. If the reconnect happens more than 24 hours after the original crash (not uncommon in environments where nsqd failures are noticed and remediated during business hours, but originally occurred on a weekend), Stripe’s idempotency cache has expired and creates ch_B even if the same key were used.

# Python pynsq — connection-scoped UUID in idempotency key (unsafe)
import uuid
import hashlib
import nsq
import stripe
import logging

class BillingConsumer:
    def __init__(self, nsqd_tcp_address: str, topic: str, channel: str):
        self.nsqd_tcp_address = nsqd_tcp_address
        self.topic = topic
        self.channel = channel
        # Unsafe: conn_uuid generated once per consumer instance startup.
        # Survives reconnects to the SAME nsqd if the consumer process stays alive.
        # Changes if the consumer process is restarted (OOM kill, pod eviction,
        # rolling deploy) — any process restart generates a new conn_uuid.
        self.conn_uuid = str(uuid.uuid4())
        self.reader = None

    def start(self):
        self.reader = nsq.Reader(
            message_handler=self.handle_message,
            nsqd_tcp_addresses=[self.nsqd_tcp_address],
            topic=self.topic,
            channel=self.channel,
            max_in_flight=5,
        )
        nsq.run()

    def handle_message(self, message):
        import json
        try:
            event = json.loads(message.body)
            customer_id = event["customer_id"]
            billing_period = event["billing_period"]
            amount_cents = event["amount_cents"]
            stripe_customer_id = event["stripe_customer_id"]

            # Unsafe: self.conn_uuid changes if this consumer process restarts.
            # nsqd disk-persisted messages redelivered after nsqd restart are
            # processed with a NEW conn_uuid on process reconnect → different Stripe key.
            #
            # Even without nsqd restart: if THIS consumer process is restarted
            # (OOM kill, pod eviction, SIGKILL from orchestrator), the next process
            # has conn_uuid = str(uuid.uuid4()) → different value → different key → ch_B.
            idempotency_key = hashlib.sha256(
                f"{self.conn_uuid}:{customer_id}:{billing_period}".encode()
            ).hexdigest()[:32]

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

            # If process is killed here (Stripe succeeded, charge created as ch_A,
            # but message.finish() never called) — next process receives the
            # requeued message with a new conn_uuid → idempotency_key is different → ch_B.
            message.finish()
        except Exception as e:
            logging.error(f"Billing error: {e}")
            message.requeue()


# Variant: conn_uuid generated per-connection rather than per-process:
#
# class BillingConsumerV2:
#     def on_connect(self, conn):
#         # UUID generated fresh on EVERY nsqd connection (re)establishment.
#         # Survives neither nsqd restart nor consumer process restart.
#         self.conn_uuid = str(uuid.uuid4())
#
# In this variant, a simple TCP keepalive timeout (default: 60s idle in many
# environments) triggers a reconnect and a new conn_uuid — even without any
# nsqd restart or process crash. A billing message whose processing spans
# 60+ seconds of Stripe latency plus DB write time can trigger a reconnect
# during processing, generating a new conn_uuid before message.finish() is sent.

A second variant that appears in production: developers use the nsqd TCP address itself (nsqd_tcp_addr — e.g., "nsqd-host-1:4150") as a component of the idempotency key, reasoning that each nsqd instance is a distinct “delivery source.” In a static NSQ deployment where nsqd daemons always bind to the same hostname and port, this produces a stable key. In Kubernetes deployments where nsqd runs as a StatefulSet pod with a stable hostname but may be rescheduled to a different node, the address is stable in DNS but the pod’s in-memory message state is not. More importantly, a billing message that is requeued and re-delivered from the same nsqd address (because the consumer crashed but nsqd survived) is processed with the same address-derived key, meaning Stripe’s 24-hour idempotency cache absorbs the retry — false safety that disappears when the 24-hour window expires. The content-hash key derived from stable business fields is stable across all of these scenarios without requiring any knowledge of the NSQ topology.

Failure mode 3: pynsq Tornado IOLoop blocking on synchronous stripe.Charge.create() — the single-threaded IOLoop cannot send TOUCH keepalives while blocked, nsqd requeues the message after msg-timeout expires during the Stripe call, the belated Finish() is ignored, and the requeued message creates ch_B

pynsq — the official Python NSQ client — is built on Tornado’s IOLoop (versions up to and including pynsq 0.9.x use Tornado 5 or 6’s IOLoop). The IOLoop is a single-threaded event loop: it processes one callback or coroutine at a time, interleaved with I/O events from sockets. This architecture works well for I/O-bound workloads where handlers yield back to the IOLoop quickly. It creates a critical failure mode for billing handlers that call synchronous Python functions that block the calling thread — including the synchronous Stripe Python SDK’s stripe.Charge.create().

NSQ’s in-flight timeout mechanism requires the consumer to send protocol commands within the msg-timeout window (default: 60 seconds). If a message will take longer than msg-timeout to process, the consumer must send a TOUCH command to extend the timeout. Both TOUCH and FIN are sent by writing bytes to the nsqd TCP connection — operations that go through the Tornado IOLoop’s write buffer. When a message handler blocks the IOLoop, the IOLoop cannot write anything to any socket. Not TOUCH commands. Not FIN commands. Not heartbeat NOP responses. NSQ’s server-side heartbeat (default: 30-second interval) expects a NOP response from the consumer; if the response does not arrive within the heartbeat window, nsqd closes the connection.

The failure sequence: pynsq Reader receives billing message M from nsqd. The message handler calls stripe.Charge.create() synchronously. The Stripe SDK makes an HTTPS request to api.stripe.com. This is a blocking network call that takes 200–800ms under normal conditions and up to 30 seconds under Stripe API latency events or slow network conditions. The IOLoop is blocked for the duration of this call. If the Stripe call takes more than 60 seconds (Stripe SDK’s default timeout) or if the consumer is handling multiple messages concurrently (pynsq’s max_in_flight > 1) and the cumulative processing time for all in-flight messages exhausts the IOLoop’s I/O bandwidth — nsqd’s msg-timeout fires. nsqd marks M as timed out, increments Attempts to 2, and makes M available for redelivery. The blocking Stripe call eventually returns: ch_A is created. The handler calls message.finish(). pynsq writes a FIN [msg_id] command to nsqd. nsqd processes the FIN, but M is no longer in the in-flight set — it was requeued when msg-timeout fired. nsqd returns E_FIN_FAILED Invalid Message ID. Depending on the nsqd version and pynsq error handling, this may close the connection or be silently swallowed. Message M is redelivered with Attempts=2 to the next available consumer, which calls stripe.charges.create() and creates ch_B.

# Python pynsq — synchronous Stripe call blocks the IOLoop (unsafe)
import nsq
import stripe
import hashlib
import json
import logging

def handle_billing_message(message):
    try:
        event = json.loads(message.body)
        customer_id = event["customer_id"]
        billing_period = event["billing_period"]

        idempotency_key = hashlib.sha256(
            f"{customer_id}:{billing_period}:nsq-billing".encode()
        ).hexdigest()[:32]

        # Unsafe: stripe.Charge.create() is a synchronous blocking call.
        # It blocks the Tornado IOLoop for up to 30 seconds (SDK default timeout).
        # While blocked:
        #   - nsqd cannot receive TOUCH keepalives → msg-timeout fires → M requeued
        #   - nsqd cannot receive heartbeat NOP response → connection may be closed
        #   - Other in-flight messages (max_in_flight > 1) cannot be FINished
        # If Stripe takes > msg-timeout (60s default), ch_A is created but nsqd
        # has already requeued M. The FIN below hits E_FIN_FAILED. ch_B is created
        # when M is redelivered to the next consumer.
        charge = stripe.Charge.create(
            amount=event["amount_cents"],
            currency="usd",
            customer=event["stripe_customer_id"],
            idempotency_key=idempotency_key,
        )

        logging.info(f"Charged {customer_id}: {charge.id}")
        message.finish()  # May arrive as E_FIN_FAILED if msg-timeout already fired
    except Exception as e:
        logging.error(f"Error: {e}")
        message.requeue(delay=30)


# Unsafe reader — synchronous handler in IOLoop thread
reader = nsq.Reader(
    message_handler=handle_billing_message,
    nsqd_tcp_addresses=["127.0.0.1:4150"],
    topic="billing-events",
    channel="stripe-processor",
    max_in_flight=5,  # Up to 5 messages in-flight simultaneously.
    # With max_in_flight=5, all 5 handlers could be blocked simultaneously
    # during a Stripe API latency event, timing out all 5 messages at once.
)


# Safe alternative: run Stripe call in a thread executor, yield back to IOLoop

import tornado.ioloop
import concurrent.futures

_executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)

def safe_handle_billing_message(message):
    # Do NOT call tornado.ioloop.IOLoop.current().run_sync() — that's for tests.
    # Use IOLoop.run_in_executor() to offload blocking Stripe call.
    loop = tornado.ioloop.IOLoop.current()

    async def _process():
        event = json.loads(message.body)
        customer_id = event["customer_id"]
        billing_period = event["billing_period"]

        idempotency_key = hashlib.sha256(
            f"{customer_id}:{billing_period}:nsq-billing".encode()
        ).hexdigest()[:32]

        # Safe: run blocking Stripe call in thread pool executor.
        # IOLoop continues processing I/O events (TOUCH, FIN, NOP heartbeats)
        # while the Stripe call runs in a worker thread.
        charge = await loop.run_in_executor(
            _executor,
            lambda: stripe.Charge.create(
                amount=event["amount_cents"],
                currency="usd",
                customer=event["stripe_customer_id"],
                idempotency_key=idempotency_key,
            )
        )
        logging.info(f"Charged {customer_id}: {charge.id}")
        message.finish()

    loop.add_callback(loop.run_sync, _process)

The severity of this failure mode scales with max_in_flight. A pynsq Reader with max_in_flight=10 can have 10 messages in-flight concurrently. If all 10 handlers are blocked simultaneously on Stripe calls during a Stripe API latency event (common when Stripe is under load), all 10 messages’ msg-timeout timers fire at the same time, all 10 messages are requeued, and all 10 consumers that receive the requeued messages call stripe.charges.create() again — 10 duplicate charges from a single latency event. This does not happen in go-nsq: go-nsq processes each message in a dedicated goroutine. One goroutine blocking on a Stripe call does not prevent other goroutines from calling m.Touch() or m.Finish(), and it does not prevent go-nsq’s heartbeat goroutine from responding to nsqd’s heartbeat with a NOP. The IOLoop-blocking failure mode is specific to pynsq’s Tornado event loop and the synchronous Stripe SDK.

A related variant occurs when the pynsq handler uses tornado.gen.sleep() or asyncio.sleep() for retry backoff between Stripe retries. These sleep calls yield back to the IOLoop (correct), but if the developer uses a sleep duration longer than the remaining msg-timeout window, the message times out during the sleep and is requeued even though no actual work is happening. The safe pattern in all cases: run the Stripe call in a ThreadPoolExecutor via IOLoop.run_in_executor(), keep the coroutine itself non-blocking (yield points only via await), and set msg-timeout in nsqd’s configuration to be longer than the maximum expected Stripe call duration plus the maximum expected database write time.

What doesn’t prevent duplicate Stripe charges in NSQ

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

Three changes close all three failure modes without modifying NSQ topic/channel configuration, nsqd deployment topology, or consumer concurrency settings.

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 NSQ delivery path: first delivery, crash-recovery requeue (new Attempts value), consumer process restart (new connection UUID), and msg-timeout redelivery from blocked IOLoop. The billing event payload’s business fields — customer ID, billing period, and a service-level namespace — satisfy all conditions. NSQ delivery metadata — Attempts counter, connection UUID, nsqd TCP address, per-handler invocation timestamp, or any value generated after the message is received — does not.

// Go go-nsq — safe content-hash idempotency key and pre-flight PostgreSQL check
package main

import (
    "context"
    "crypto/sha256"
    "database/sql"
    "encoding/json"
    "fmt"

    "github.com/nsqio/go-nsq"
    "github.com/stripe/stripe-go/v76"
    "github.com/stripe/stripe-go/v76/charge"
    _ "github.com/lib/pq"
)

type SafeBillingHandler struct {
    db *sql.DB
}

func contentHashKey(customerID, billingPeriod string) string {
    // Inputs: stable business fields only.
    // Excluded (all NSQ delivery metadata):
    //   - m.Attempts (starts at 1, increments on every requeue including crash-recovery)
    //   - connection UUID (changes on consumer process restart or nsqd reconnect)
    //   - nsqd TCP address (changes in dynamic nsqd cluster topologies)
    //   - time.Now().UnixNano() inside HandleMessage (per-invocation, changes every call)
    //   - any UUID generated inside HandleMessage (per-invocation, not stable)
    raw := fmt.Sprintf("%s:%s:nsq-billing", customerID, billingPeriod)
    sum := sha256.Sum256([]byte(raw))
    return fmt.Sprintf("%x", sum)[:32]
}

func (h *SafeBillingHandler) HandleMessage(m *nsq.Message) error {
    var event BillingEvent
    if err := json.Unmarshal(m.Body, &event); err != nil {
        return err
    }

    // Safe: content-hash key — same value on every delivery of this billing event.
    // m.Attempts=1, m.Attempts=2, m.Attempts=3 all produce the same key.
    idempotencyKey := contentHashKey(event.CustomerID, event.BillingPeriod)

    // Pre-flight check: INSERT into external PostgreSQL before calling Stripe.
    // ON CONFLICT DO NOTHING: if (customer_id, billing_period) already exists
    // (because a previous delivery called stripe.Charges.New() before the crash),
    // RowsAffected() = 0 and we skip Stripe entirely.
    // PostgreSQL survives consumer process restarts, nsqd restarts, and reconnects.
    res, err := h.db.ExecContext(context.Background(),
        `INSERT INTO billing_records (customer_id, billing_period, idempotency_key)
         VALUES ($1, $2, $3) ON CONFLICT (customer_id, billing_period) DO NOTHING`,
        event.CustomerID, event.BillingPeriod, idempotencyKey,
    )
    if err != nil {
        return err // retry on transient DB error
    }

    rowsAffected, _ := res.RowsAffected()
    if rowsAffected == 0 {
        // Pre-flight row already exists: a prior delivery called Stripe.
        // Skip Stripe and finish the message — billing already processed.
        m.Finish()
        return nil
    }

    // Write-before-call ordering:
    // - billing_records row is committed BEFORE stripe.Charges.New().
    // - If handler is killed between DB commit and Stripe call:
    //     Stripe was never called (customer not charged — correct).
    //     Pre-flight row exists on next delivery → Stripe skipped.
    // - If handler is killed between Stripe call and m.Finish():
    //     ch_A exists in Stripe.
    //     Pre-flight row exists on next delivery → Stripe skipped → no ch_B.
    params := &stripe.ChargeParams{
        Amount:   stripe.Int64(event.AmountCents),
        Currency: stripe.String(string(stripe.CurrencyUSD)),
        Customer: stripe.String(event.StripeCustomerID),
    }
    params.IdempotencyKey = stripe.String(idempotencyKey)

    ch, err := charge.New(params)
    if err != nil {
        return err
    }

    // Update billing_records with the Stripe charge ID (best-effort — pre-flight row exists).
    h.db.ExecContext(context.Background(),
        `UPDATE billing_records SET charge_id = $1 WHERE customer_id = $2 AND billing_period = $3`,
        ch.ID, event.CustomerID, event.BillingPeriod,
    )

    // If process is killed here — ch_A exists, pre-flight row exists.
    // NSQ requeue with m.Attempts+1 → same content-hash key → pre-flight row found → no ch_B.
    m.Finish()
    return nil
}

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

The content-hash idempotency key closes failure modes 1 and 2 within Stripe’s 24-hour idempotency window: Stripe returns ch_A for every call using the same key within 24 hours, regardless of which NSQ delivery triggered the call. Failure modes 1 and 2 can occur outside the 24-hour window in environments where NSQ messages queue up behind slow consumers, nsqd restarts are delayed, or billing runs span multiple days (common in subscription billing systems that process large customer cohorts over 48–72 hours).

The pre-flight PostgreSQL check closes all three failure modes regardless of the 24-hour window. 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 rowsAffected. If rowsAffected == 1 (row inserted), the billing event is new and the worker calls Stripe. If rowsAffected == 0 (conflict on unique constraint), a prior delivery already initiated Stripe processing and the worker skips the call and FINishes the message. The PostgreSQL table is external to the NSQ consumer process: it survives consumer process restarts (which change the connection UUID), nsqd restarts (which change the nsqd-side message state), and IOLoop resets (which may close and reopen the nsqd connection). The UNIQUE (customer_id, billing_period) constraint is the durable deduplication layer that NSQ’s at-least-once delivery model does not provide on its own.

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 an NSQ crash-recovery requeue event. If PostgreSQL is transiently unreachable (connection pool exhausted, rolling maintenance, network partition between the billing consumer and the database), the pre-flight check cannot execute. Without the vault cap, stripe.charges.create() might proceed with a different NSQ-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 total beyond the 110% cap. The vault cap acts at the Stripe API boundary, downstream of both the NSQ transport and the pre-flight check, providing defense in depth when both upstream layers fail simultaneously — including during the thundering-herd requeue event from a pynsq IOLoop block that times out all in-flight messages at once.

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

NSQ topic pause and resume — messages accumulate during pause and are replayed with elevated Attempts counts after resume

NSQ supports pausing a topic or channel via nsqadmin or the HTTP API (POST /topic/pause). During a pause, nsqd queues messages in memory (and on disk when --mem-queue-size is reached) but does not deliver them to consumers. This is used during deployments, maintenance windows, or when consumers need to be drained. When the topic is resumed, nsqd delivers the accumulated messages. If some of those messages were already in-flight when the pause was initiated — received by consumers but not yet FINished — those messages were requeued at pause time and their Attempts counter was incremented. When resume fires, those messages arrive with Attempts=2 or higher. A consumer using m.Attempts in the Stripe key processes them as if they were fresh retries of failed charges and creates ch_B for each one. The content-hash key is stable across the pause/resume boundary because it does not reference Attempts; the pre-flight check prevents duplicate charges even for billing events whose original delivery charged Stripe before the pause.

NSQ consumer group horizontal scaling — new consumer pod joins the channel during a billing run and receives messages requeued from the previous pod

NSQ channels load-balance messages across all connected consumers (within a channel, each message goes to exactly one consumer at a time). When a consumer pod is OOM-killed during a billing run, its in-flight messages time out and are requeued. A new pod spins up to replace it, connects to nsqlookupd, discovers the same nsqd instances, and subscribes to the same channel. The requeued billing messages are delivered to the new pod. If the new pod generates a connection UUID at startup and uses it in the Stripe idempotency key, the key differs from the old pod’s key — even if the old pod had already called Stripe successfully. The new pod creates ch_B. The content-hash key is stable across pod replacements because it derives from billing event payload fields that are identical in both the old and new pod’s copies of the message.

NSQ deferred messages (REQ with delay) for retry backoff — developer calls message.requeue(delay=300) after a Stripe timeout, redelivery happens after Stripe’s idempotency cache window

A billing consumer that catches stripe.error.Timeout often calls message.requeue(delay=300) (a 5-minute deferred requeue) to implement exponential backoff. A stripe.error.Timeout means the Stripe API server may or may not have processed the charge: the server received the request but the response was not received before the SDK’s timeout. If Stripe did process the request, ch_A exists. If the deferred requeue delay plus subsequent processing time pushes the retry beyond 24 hours from the original request, Stripe’s idempotency cache has expired and the retry creates ch_B even with the same content-hash key. The pre-flight PostgreSQL check closes this case: the pre-flight row was written before the first stripe.charges.create() call (write-before-call ordering), so the deferred requeue finds the row on delivery, skips Stripe, and FINishes the message regardless of whether the first Stripe call completed or was interrupted mid-network. The content-hash key closes the case where the first Stripe call did complete and the requeue happens within 24 hours.

Frequently asked questions

Can I use m.ID (the NSQ message ID) directly as the Stripe idempotency key without a content-hash?

m.ID is a 16-byte value assigned by nsqd at message publication time. It is stable across all redeliveries of the same message — nsqd does not change the message ID on requeue. Used as the sole Stripe idempotency key (stripe.IdempotencyKey = hex(m.ID)), it prevents failure modes 1 and 2 within Stripe’s 24-hour idempotency window and prevents FM3 as well (same key on every delivery). The limitation: if the same billing event is published to NSQ more than once (a producer-side retry after a failed nsqd PUB command, or an upstream system that re-enqueues failed events), each publication generates a new nsqd message with a new m.ID. Two messages for the same customer and billing period have different IDs and would create two Stripe charges. The content-hash key derived from (customer_id, billing_period) is safe against both consumer-side redeliveries (same message, same ID, same content-hash) and producer-side re-publishes (different message, different ID, but same content-hash because the payload fields are identical). The pre-flight PostgreSQL check provides additional protection in both cases beyond the 24-hour Stripe window.

Does go-nsq’s m.Touch() prevent failure mode 3 in Python?

In go-nsq, calling m.Touch() from within a goroutine extends the message’s in-flight timeout by sending a TOUCH [msg_id] command to nsqd. Since go-nsq processes each message in a dedicated goroutine, one goroutine can call m.Touch() in a loop while another goroutine’s Stripe call is blocking — they run concurrently. This prevents FM3 in go-nsq. In pynsq, calling the equivalent message.touch() during a blocked IOLoop handler is impossible: the handler itself is running synchronously in the IOLoop thread and cannot yield for the touch write. The only safe approach is to run the Stripe call in a ThreadPoolExecutor via loop.run_in_executor(), which frees the IOLoop to process touch timers (if any), heartbeat responses, and FIN commands while the blocking Stripe SDK call runs in a worker thread.

If stripe.charges.create() returns stripe.error.IdempotencyError, does that mean ch_B was not created?

Stripe raises stripe.error.IdempotencyError when a request is made with the same idempotency key as a prior request but with different request parameters (amount, currency, or customer). This error is specific to parameter mismatches — it does not indicate a duplicate charge attempt with matching parameters. When the content-hash key is used consistently and the billing event payload is stable across deliveries, the parameters are identical on every retry. Stripe returns the same charge object (with the original ch_A charge ID) on every call with a matching key and matching parameters within 24 hours. IdempotencyError should not appear when the content-hash key is derived from the same billing period and customer ID fields that are also sent as the amount and customer Stripe parameters. If IdempotencyError does appear, it indicates that the amount or Stripe customer ID changed between deliveries — a data integrity problem in the billing event payload, not an NSQ delivery problem.

What happens if nsqd’s --mem-queue-size=0 and the disk persistence file is corrupted on restart?

With --mem-queue-size=0, all messages are immediately flushed to disk (--data-path). If the disk persistence file is corrupted on nsqd restart (truncated write, filesystem corruption, SIGKILL during the sync write), nsqd logs a parse error and may discard the corrupted segment. Billing messages in the corrupted segment are permanently lost from the NSQ queue — not redelivered, not requeued. This is a data-loss scenario, not a duplicate-charge scenario. The pre-flight PostgreSQL check actually helps here: messages that were processed (pre-flight row written) but whose Stripe call succeeded before the corruption are correctly recorded in billing_records. Messages whose pre-flight row was not written before nsqd crashed are lost from NSQ, meaning the customer is never charged (ch_A was never created). The correct remediation in this scenario is a reconciliation pass that compares billing_records against the expected customer billing list and re-publishes any messages that have no pre-flight row, using a fresh content-hash idempotency key derived from the same (customer_id, billing_period) fields.