Deduplicating Streamed Billing Events

The specific bottleneck this page solves is redelivery: Pub/Sub — and SNS/SQS behave the same way — guarantee at-least-once delivery, never exactly-once, so a budget or cost event you already processed will arrive again, sometimes minutes later, sometimes hours later. A consumer that treats every delivery as a new event double-counts spend in a running total, fires a duplicate PagerDuty page for the same overrun, or double-applies a credit reversal. This is not an edge case to be handled with a try/except; it is the baseline delivery contract of the queue, and every consumer sitting behind Streaming Billing Data with Pub/Sub has to assume redelivery on every message. The fix lives at the application layer: a content-derived dedup key checked against a TTL-bounded store before any billing side effect runs.

# Root Cause & Failure Modes

Redelivery is not a bug in Pub/Sub — it is the deliberate trade-off behind the at-least-once guarantee, and it has two distinct, quantifiable causes:

• Ack deadline expiry. Every subscription has an ack deadline, 10 seconds by default and extendable via modifyAckDeadline to a hard ceiling of 600 seconds (10 minutes). If your handler is still writing to the cost ledger when that window closes — a slow downstream write, a garbage-collection pause, a thread pool that is briefly saturated — Pub/Sub assumes the consumer died and redelivers the same message to another worker. The original worker often finishes and acks a moment later, so both copies get “successfully” processed.
• Publisher-side retries on ambiguous failures. If a publish RPC times out or the connection drops after the broker committed the message but before the acknowledgment reaches the publisher, the client cannot tell success from failure and retries. The topic now holds two physically distinct messages carrying the same logical billing event, each with its own message_id — deduplicating on message_id does nothing here, because there is no shared identifier to dedupe against.
• Exactly-once delivery does not remove the requirement. Pub/Sub’s exactly-once feature suppresses duplicate deliveries only within a single region, only while the subscriber acks within the deadline, and does nothing about the second failure mode above (duplicate publishes are a distinct message by design). SNS/SQS has the equivalent gap: standard queues are explicitly at-least-once, and even SQS FIFO’s deduplication window is capped at 5 minutes — well under the redelivery latency a backlog can introduce.

The only structural fix is to make the consumer idempotent: derive a deterministic key from the event’s content, not its transport metadata, and refuse to apply any key twice within the retention window.

# Production Pipeline Architecture

The pattern is a four-phase gate sitting in front of the actual billing write, and it is deliberately decoupled from whichever ingestion path — Streaming Billing Data with Pub/Sub or the batch alternative compared in BigQuery Export vs Pub/Sub Streaming for Billing — produced the message.

1. Deserialize. Parse the message body into a typed BillingEvent. Malformed payloads are acked immediately (never nacked) so a poison message cannot loop forever against the redelivery policy.
2. Hash. Compute a SHA-256 digest over the fields that define billing identity — account, SKU, usage window, cost type, amount, currency — explicitly excluding message_id and publish_time, since those change on every redelivery of the same logical event.
3. Claim. Atomically test-and-set that hash against a TTL store. A Redis SET ... NX PX or a DynamoDB PutItem with ConditionExpression=attribute_not_exists(dedup_key) both give a single atomic winner even when two workers race on the same redelivered message; the loser’s write is rejected, not silently overwritten.
4. Apply and ack. Only the claim winner writes to the cost ledger, then acks. If the write fails after the claim succeeds, nack explicitly rather than waiting out the ack deadline — the claim already exists, so the redelivered retry is dropped as a duplicate unless you also expire the claim on failure.

The TTL matters as much as the atomicity: set it to cover the subscription’s message retention (7 days by default on Pub/Sub, extendable to 31) plus a safety margin, not to processing latency. A dedup key that expires before a slow-draining backlog catches up lets a genuine duplicate back through. The same content-hash-plus-TTL discipline generalizes to the batch-side pagination problem in Azure Cost API Pagination and Deduplication Guide — there the duplicate source is overlapping nextLink pages instead of queue redelivery, but the composite-key gate is the same idea applied before the sink write.

# Step-by-Step Python Implementation

The module below is a runnable Pub/Sub subscriber with a pluggable dedup backend. It ships both a Redis and a DynamoDB implementation of the same claim() contract, computes the content hash off the billing-identity fields only, and nacks explicitly on write failure so the claim already recorded does not silently swallow a legitimate retry.

import hashlib
import json
import logging
import os
from abc import ABC, abstractmethod
from concurrent.futures import TimeoutError as FutureTimeoutError
from dataclasses import dataclass

import boto3
import redis
from google.api_core.exceptions import GoogleAPIError
from google.cloud import pubsub_v1
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

PROJECT_ID = os.getenv("GCP_PROJECT_ID")
SUBSCRIPTION_ID = os.getenv("PUBSUB_SUBSCRIPTION", "billing-events-sub")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
DEDUP_TTL_SECONDS = 31 * 24 * 3600  # covers max Pub/Sub retention (31d) plus margin


@dataclass(frozen=True)
class BillingEvent:
    account_id: str
    sku_id: str
    usage_start_time: str
    cost_type: str
    amount: float
    currency: str

    def content_hash(self) -> str:
        """SHA-256 over billing identity only -- never message_id/publish_time,
        which change on every redelivery of the same logical event."""
        canonical = json.dumps(
            {"account_id": self.account_id, "sku_id": self.sku_id,
             "usage_start_time": self.usage_start_time, "cost_type": self.cost_type,
             "amount": round(self.amount, 6), "currency": self.currency},
            sort_keys=True,
        )
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class DedupStore(ABC):
    @abstractmethod
    def claim(self, dedup_key: str) -> bool:
        """Return True only for the first caller to claim this key."""


class RedisDedupStore(DedupStore):
    def __init__(self, client: redis.Redis, ttl_seconds: int = DEDUP_TTL_SECONDS):
        self._client, self._ttl = client, ttl_seconds

    @retry(reraise=True, stop=stop_after_attempt(3),
           wait=wait_exponential(multiplier=0.5, max=4),
           retry=retry_if_exception_type(redis.RedisError))
    def claim(self, dedup_key: str) -> bool:
        # SET NX PX is a single atomic op -- two racing workers can never both win.
        claimed = self._client.set(f"billing:dedup:{dedup_key}", "1", nx=True, px=self._ttl * 1000)
        return bool(claimed)


class DynamoDBDedupStore(DedupStore):
    def __init__(self, table_name: str, ttl_seconds: int = DEDUP_TTL_SECONDS):
        self._table = boto3.resource("dynamodb").Table(table_name)
        self._ttl = ttl_seconds

    def claim(self, dedup_key: str) -> bool:
        import time
        try:
            self._table.put_item(
                Item={"dedup_key": dedup_key, "expires_at": int(time.time()) + self._ttl},
                ConditionExpression="attribute_not_exists(dedup_key)",
            )
            return True
        except self._table.meta.client.exceptions.ConditionalCheckFailedException:
            return False  # already claimed by a prior delivery


def apply_billing_event(event: BillingEvent) -> None:
    """Idempotent sink write -- placeholder for the real cost-ledger upsert."""
    logger.debug("Writing to cost ledger: %s", event)


def handle_message(message: "pubsub_v1.subscriber.message.Message", store: DedupStore) -> None:
    try:
        event = BillingEvent(**json.loads(message.data.decode("utf-8")))
    except (json.JSONDecodeError, TypeError) as exc:
        logger.error("Malformed billing event, acking to drop it: %s", exc)
        message.ack()  # a poison message must never be nacked, or it loops forever
        return

    dedup_key = event.content_hash()
    if not store.claim(dedup_key):
        logger.info("Duplicate dropped (key=%s..., sku=%s)", dedup_key[:12], event.sku_id)
        message.ack()
        return

    try:
        apply_billing_event(event)
        message.ack()
        logger.info("Applied (key=%s..., amount=%.4f %s)", dedup_key[:12], event.amount, event.currency)
    except Exception:
        # Claim already recorded; nack forces immediate redelivery instead of
        # waiting the full ack deadline, and the retry re-enters this same path.
        message.nack()
        logger.exception("Apply failed; nacked for redelivery (key=%s...)", dedup_key[:12])


def main() -> None:
    store: DedupStore = RedisDedupStore(redis.Redis.from_url(REDIS_URL, socket_timeout=2.0))
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(PROJECT_ID, SUBSCRIPTION_ID)
    flow_control = pubsub_v1.types.FlowControl(max_messages=200)

    future = subscriber.subscribe(
        subscription_path,
        callback=lambda message: handle_message(message, store),
        flow_control=flow_control,
    )
    logger.info("Listening on %s ...", subscription_path)
    with subscriber:
        try:
            future.result(timeout=None)
        except (FutureTimeoutError, GoogleAPIError, KeyboardInterrupt):
            future.cancel()
            future.result()


if __name__ == "__main__":
    main()

# Verification & Testing

• Replay assertion. Publish the identical payload twice in a test topic and drain the subscription. Assert apply_billing_event was invoked exactly once and the second claim() call returned False — this is the core contract and should be a unit test, not something discovered in production.
• Race assertion. Spin up two threads calling claim() concurrently with the same dedup_key against a real Redis or DynamoDB instance (not a mock — the atomicity is the thing under test). Exactly one must return True.
• TTL boundary check. Set DEDUP_TTL_SECONDS short in a test environment, wait past expiry, and confirm a “duplicate” then claims successfully again — this proves the store expires rather than growing unbounded, and that your production TTL is deliberately longer than retention, not shorter.
• Poison-message dry run. Publish a payload missing a required field and confirm handle_message acks (not nacks) it, and that the failure is logged with enough context to replay manually.

# Common Pitfalls Checklist

• Hashing the raw message bytes instead of normalized fields. JSON key ordering or float formatting differences between a redelivery and a retried publish produce two different hashes for the same event — always serialize with sort_keys=True and rounded numerics before hashing.
• Using message_id as the dedup key. Pub/Sub assigns a new message_id on every redelivery and on every retried publish, so this defeats deduplication entirely — hash the payload content instead.
• Setting the TTL to processing latency instead of message retention. A short TTL lets a duplicate that arrives after a backlog drain slip through the closed claim window — size the TTL to subscription retention plus margin.
• Acking before the write succeeds. Acking early loses the message if the ledger write then fails, with no redelivery to recover it — ack only after apply_billing_event returns.
• Nacking malformed payloads. A message that can never deserialize will redeliver forever under a nack and burn through the max-delivery-attempts dead-letter threshold uselessly — ack and log poison messages instead.

# Frequently Asked Questions

# Related

• Streaming Billing Data with Pub/Sub — the ingestion engine this dedup gate sits in front of, covering subscription setup and ordering keys.
• BigQuery Export vs Pub/Sub Streaming for Billing — when a periodic export beats a real-time stream, and where redelivery risk does or doesn’t apply.
• Azure Cost API Pagination and Deduplication Guide — the same composite-key dedup discipline applied to overlapping REST pagination instead of queue redelivery.
• Building Fault-Tolerant Billing Ingestion with Celery — retry and idempotency patterns for the async task layer that often sits downstream of this consumer.
• Cloud Billing Data Ingestion & Parsing — the broader acquisition-to-persistence pipeline this consumer’s output feeds into.

Up: Streaming Billing Data with Pub/Sub · Home: Cloud Cost Optimization & FinOps Automation