Amazon Kinesis Firehose and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Amazon Kinesis Data Firehose (now Amazon Data Firehose) is AWS’s fully managed delivery service for streaming data to S3, Redshift, OpenSearch, Splunk, and HTTP endpoints. When a billing pipeline reads usage events from a Firehose-populated destination and calls stripe.charges.create() for each, three Firehose-specific behaviors — S3 at-least-once delivery retry, Redshift COPY retry after connection loss, and HTTP endpoint request body timestamp drift — introduce Stripe double-charge failure modes that visibility tuning and retry backoff alone cannot prevent.

This post covers those three failure modes with Python and boto3 code, content-hash idempotency keys that exclude all Firehose delivery metadata, per-billing-period vault keys via a spend-cap proxy, and a pre-flight PostgreSQL database check — the two-layer governance pattern that closes all three without changing Firehose configuration, destination type, or buffer flush settings.

Failure mode 1: S3 destination — Firehose retries a delivery after the S3 PutObject succeeds but the acknowledgment is lost in transit; a second S3 object is created with the same records; S3 event notification triggers the billing Lambda twice; both invocations call stripe.charges.create() for the same customers

Firehose buffers incoming records until it reaches the configured flush threshold — buffer size (1–128 MB, default 5 MB) or buffer interval (60–900 seconds, default 300 seconds) — and then flushes by calling S3 PutObject. Each flushed S3 object gets a key constructed from a timestamp prefix and a UUID suffix: for example, billing/2026/07/25/01/firehose-billing-stream-1-2026-07-25-01-00-00-a1b2c3d4. The timestamp prefix encodes the delivery time; the UUID suffix is generated by Firehose for each delivery attempt.

The failure scenario: Firehose calls PutObject for a buffer containing 500 billing event records. S3 writes the object successfully and returns HTTP 200. The TCP connection between Firehose and the S3 endpoint drops immediately after S3 sends the response — due to a VPC endpoint transient failure, a network load balancer health check race, or a brief S3 service interruption — before Firehose receives the HTTP 200. Firehose’s delivery layer interprets the missing response as a failed delivery. The retry kicks in according to the configured retry duration (default 24 hours for S3 destination). Firehose generates a new UUID suffix and calls PutObject again with the same 500 buffered records: billing/2026/07/25/01/firehose-billing-stream-1-2026-07-25-01-00-03-e5f6g7h8. This delivery succeeds (Firehose receives the HTTP 200).

Both objects now exist in S3. An S3 event notification subscribed to s3:ObjectCreated:Put on the billing prefix fires for each. If the notification delivers to an SQS queue, two messages enter the queue independently. A billing Lambda reading from SQS invokes twice — once per object. Each invocation reads its S3 object, iterates over 500 billing records, and calls stripe.charges.create() for each customer:

import json
import boto3
import stripe

s3 = boto3.client('s3')

def lambda_handler(event, context):
    for sqs_record in event['Records']:
        s3_event = json.loads(sqs_record['body'])
        for s3_record in s3_event['Records']:
            bucket = s3_record['s3']['bucket']['name']
            key = s3_record['s3']['object']['key']

            obj = s3.get_object(Bucket=bucket, Key=key)
            lines = obj['Body'].read().decode('utf-8').strip().split('\n')

            for line in lines:
                billing = json.loads(line)
                customer_id = billing['customer_id']
                billing_period = billing['billing_period']

                # UNSAFE: idempotency key derived from the S3 key.
                # Firehose retry writes the same records to a different S3
                # key (new UUID suffix). The second Lambda invocation
                # produces a different idempotency_key for the same
                # customer and billing period. Stripe creates ch_B.
                idempotency_key = f"{key}:{customer_id}:{billing_period}"

                charge = stripe.Charge.create(
                    amount=billing['amount_cents'],
                    currency='usd',
                    customer=billing['stripe_customer_id'],
                    description=f'Usage billing for {billing_period}',
                    idempotency_key=idempotency_key,
                )

The S3 key includes the UUID suffix generated by Firehose per delivery attempt. The original delivery’s key ends in -a1b2c3d4; the retry key ends in -e5f6g7h8. Two idempotency keys — two distinct requests from Stripe’s perspective — two charges.

This failure mode is insidious for two reasons. First, Firehose delivery metrics report both deliveries as successful (the first PutObject succeeded at S3; the second PutObject succeeded and was acknowledged). The CloudWatch DeliveryToS3.Success metric does not distinguish between original deliveries and retries. Second, the two S3 objects have different keys, different ETag values, and appear to S3 as independent objects. A routine audit that checks for duplicate S3 objects by key will not find them; you need to hash the object content and look for content duplicates across the billing prefix.

Failure mode 2: Redshift destination — Firehose COPY command completes but the Data API connection drops before Firehose receives the FINISHED status; retry COPY inserts duplicate rows; billing job processes both rows and charges the same customers twice

For Redshift destination, Firehose uses a two-step delivery process. First, it writes the buffered records to an S3 staging prefix. Second, it executes a Redshift COPY command via the Redshift Data API (or a JDBC connection for legacy clusters): COPY billing_events FROM 's3://staging-bucket/2026/07/25/01/firehose-1-2026-07-25-01-00-00-a1b2c3d4' CREDENTIALS '...' FORMAT AS JSON 'auto'. Firehose polls the Data API for the statement status (SUBMITTED, PICKED, STARTED, FINISHED, FAILED, ABORTED).

The failure scenario: the COPY command finishes successfully — Redshift inserts all rows from the staging file into the billing_events table — and the Data API transitions the statement status to FINISHED. Before Firehose polls and reads the FINISHED status, the Data API connection is interrupted: a transient Redshift endpoint error, a network partition between the Firehose service endpoint and the Redshift cluster, or a brief Data API service disruption. Firehose times out waiting for the status poll response and treats the delivery as failed.

The retry duration for Redshift destination is configurable up to 7200 seconds. Within the retry window, Firehose writes a new S3 staging file containing the same records and executes another COPY command. The COPY succeeds. The billing_events table now contains duplicate rows:

-- After a Firehose COPY retry, billing_events contains:
--
-- customer_id | billing_period | amount_cents | firehose_staging_key         | inserted_at
-- ---------------------------------------------------------------------------
-- cus_abc123  | 2026-07        | 4900         | firehose-1-...-a1b2c3d4.json | 2026-07-25 01:00:01
-- cus_def456  | 2026-07        | 9900         | firehose-1-...-a1b2c3d4.json | 2026-07-25 01:00:01
-- cus_abc123  | 2026-07        | 4900         | firehose-1-...-b2c3d4e5.json | 2026-07-25 01:03:07
-- cus_def456  | 2026-07        | 9900         | firehose-1-...-b2c3d4e5.json | 2026-07-25 01:03:07
--
-- firehose_staging_key differs between original and retry (new S3 key for retry staging file).
-- Without a UNIQUE constraint on (customer_id, billing_period), both rows exist.

-- UNSAFE billing job: processes both rows
SELECT customer_id, billing_period, amount_cents, stripe_customer_id
FROM billing_events
WHERE processed_at IS NULL
ORDER BY inserted_at;

import psycopg2
import stripe

def run_billing_job(conn):
    cur = conn.cursor()

    # UNSAFE: no deduplication of duplicate rows from Firehose COPY retry.
    # Returns 4 rows for 2 customers (2 original + 2 retry copies).
    cur.execute("""
        SELECT id, customer_id, billing_period, amount_cents, stripe_customer_id
        FROM billing_events
        WHERE processed_at IS NULL
        ORDER BY inserted_at
    """)
    rows = cur.fetchall()

    for row in rows:
        row_id, customer_id, billing_period, amount_cents, stripe_customer_id = row

        # UNSAFE: idempotency key includes row_id (PostgreSQL serial/bigserial).
        # Each duplicate row has a different row_id. The two rows for
        # cus_abc123 / 2026-07 have row_ids 1 and 3. Stripe receives two
        # distinct idempotency keys and creates ch_A and ch_B.
        idempotency_key = f"billing-{row_id}-{customer_id}-{billing_period}"

        charge = stripe.Charge.create(
            amount=amount_cents,
            currency='usd',
            customer=stripe_customer_id,
            description=f'Usage billing for {billing_period}',
            idempotency_key=idempotency_key,
        )

        cur.execute(
            "UPDATE billing_events SET processed_at = NOW() WHERE id = %s",
            (row_id,)
        )
        conn.commit()

The duplicate rows have different id values (PostgreSQL assigns a new serial for each INSERT) and different firehose_staging_key values (different S3 staging file per retry). An idempotency key built from row_id or firehose_staging_key is distinct for each row. Stripe processes both requests independently.

The failure mode is not visible in Firehose or Redshift metrics. Firehose reports both COPY executions as successful deliveries. CloudWatch DeliveryToRedshift.Success counts both. Redshift Query History shows both COPY statements with status FINISHED and rows_inserted matching the number of records in each staging file. The duplicate rows are only detectable by querying for GROUP BY customer_id, billing_period HAVING COUNT(*) > 1 on the billing_events table — and only if the operator runs that query before the billing job marks the rows as processed.

Failure mode 3: HTTP endpoint destination — Firehose delivery request body timestamp field changes on retry; billing endpoint constructs Stripe idempotency key from delivery timestamp plus record fields; different timestamp on retry produces a different idempotency key; Stripe creates ch_B for the same customer and billing period

Firehose HTTP endpoint delivery sends batches as HTTP POST requests with a JSON body:

{
  "requestId": "ed4acda5-034f-9f42-bba1-f29aea6d7d8f",
  "timestamp": 1578090901599,
  "records": [
    { "data": "eyJjdXN0b21lcl9pZCI6ICJjdXNfYWJjMTIzIiwgLi4ufQ==" },
    { "data": "eyJjdXN0b21lcl9pZCI6ICJjdXNfZGVmNDU2IiwgLi4ufQ==" }
  ]
}

The requestId is stable across retries of the same batch — it is explicitly documented by AWS as the idempotency token for HTTP endpoint delivery. The timestamp is the server-side delivery attempt timestamp in milliseconds since epoch; it is set at the moment Firehose initiates each delivery attempt. On a retry, Firehose generates a new attempt timestamp, and the timestamp field in the request body changes. The records array contains the same records as the original delivery.

The failure scenario: the billing endpoint receives the first delivery, decodes each base64 record, and calls stripe.charges.create() for each customer. It then encounters a transient error before returning HTTP 200 — a PostgreSQL connection timeout while writing audit records, a downstream monitoring webhook call that blocks the response path, or an OOM kill during response serialization. The endpoint returns HTTP 500 or times out (Firehose’s per-request timeout is configurable, with a minimum of 3 minutes). Firehose records the delivery as failed and retries within the configured retry duration (up to 7200 seconds).

The retry arrives with the same requestId and the same records array, but a new timestamp:

import json
import base64
import hashlib
import stripe
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/firehose/billing', methods=['POST'])
def firehose_billing():
    body = request.get_json()

    request_id = body['requestId']    # stable across retries
    timestamp = body['timestamp']      # CHANGES on every retry attempt
    records = body['records']

    for record in records:
        billing = json.loads(base64.b64decode(record['data']))
        customer_id = billing['customer_id']
        billing_period = billing['billing_period']

        # UNSAFE: idempotency key includes delivery timestamp.
        # Firehose sets a new timestamp on each retry attempt.
        # Original delivery: timestamp=1578090901599
        # Retry delivery:    timestamp=1578090962104
        # Two different idempotency keys → two Stripe charges.
        idempotency_key = hashlib.sha256(
            f"{timestamp}:{customer_id}:{billing_period}".encode()
        ).hexdigest()[:32]

        charge = stripe.Charge.create(
            amount=billing['amount_cents'],
            currency='usd',
            customer=billing['stripe_customer_id'],
            description=f'Usage billing for {billing_period}',
            idempotency_key=idempotency_key,
        )

    return jsonify({"requestId": request_id}), 200

The difference between original and retry timestamp in the example is 60 seconds — a realistic Firehose retry interval. The resulting SHA-256 hash is entirely different. Stripe treats both requests as independent and creates ch_A and ch_B.

A subtler variant: a developer uses requestId for batch-level deduplication (checking a Redis key or database table before processing the batch) but derives per-record Stripe idempotency keys from timestamp + customer_id + billing_period. The batch-level dedup is correct — the same requestId will not be processed twice if the Redis/database check succeeds. But if the batch-level dedup store is unavailable (Redis timeout, database outage) at the time of retry, the endpoint falls through to per-record processing, where the timestamp-based idempotency keys are different from the original delivery’s keys. Stripe creates duplicates on the retry.

The governance pattern that closes all three failure modes

1. Content-hash idempotency key from stable record-body fields only

The idempotency key must be derived exclusively from fields that are stable across all Firehose delivery retries and across all destination types:

import hashlib

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Must NOT include:
    #   - S3 object key or UUID suffix (changes on each Firehose S3 delivery retry)
    #   - Firehose request body 'timestamp' (changes on each HTTP endpoint retry)
    #   - Redshift row id / serial / sequence (assigned at INSERT time; different per row)
    #   - firehose_staging_key (different S3 staging file per Redshift COPY retry)
    #   - ApproximateArrivalTimestamp (if reading from Kinesis Data Streams upstream)
    #
    # MUST be derived from business-level record body fields only.
    raw = f"{customer_id}:{billing_period}:firehose-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

This key is identical regardless of whether the billing records arrive via an S3 object created on the first delivery attempt or the retry, via a Redshift row inserted from the first COPY or the second, or via an HTTP endpoint request with the original or retry timestamp. Stripe’s idempotency window (24 hours) ensures that a Firehose retry arriving within 24 hours of the original delivery returns the original charge object without creating a new one.

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

A content-hash idempotency key closes duplicates that occur within Stripe’s 24-hour idempotency window. Firehose S3 retry duration can be configured up to 24 hours, but delivery metadata differences (S3 key, timestamp) mean that any retry arriving after the 24-hour window creates a new charge unless a durable record exists outside Stripe. A pre-flight PostgreSQL check with a UNIQUE constraint closes the window regardless of Firehose retry timing:

import psycopg2
import stripe
import hashlib

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    raw = f"{customer_id}:{billing_period}:firehose-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def process_billing_record(conn, billing: dict, vault_key: str):
    customer_id = billing['customer_id']
    billing_period = billing['billing_period']
    amount_cents = billing['amount_cents']
    stripe_customer_id = billing['stripe_customer_id']

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    cur = conn.cursor()

    # Pre-flight check: attempt to write before calling Stripe.
    # ON CONFLICT DO NOTHING: if a row already exists for this
    # (customer_id, billing_period), the INSERT is silently skipped.
    # The 'processed' column starts as False; we update it to True
    # after Stripe confirms the charge.
    cur.execute("""
        INSERT INTO billing_records
            (customer_id, billing_period, idempotency_key, processed, inserted_at)
        VALUES (%s, %s, %s, FALSE, NOW())
        ON CONFLICT (customer_id, billing_period) DO NOTHING
        RETURNING id
    """, (customer_id, billing_period, idempotency_key))
    conn.commit()

    row = cur.fetchone()
    if row is None:
        # Another delivery (original or retry) already inserted this row.
        # The original worker either called or will call stripe.charges.create().
        # Skip this duplicate record entirely.
        return

    # Pre-flight INSERT succeeded — we own this billing period.
    # Call Stripe using the vault key (spend-capped proxy).
    try:
        charge = stripe.Charge.create(
            amount=amount_cents,
            currency='usd',
            customer=stripe_customer_id,
            description=f'Usage billing for {billing_period}',
            idempotency_key=idempotency_key,
            api_key=vault_key,   # scoped vault key via Keybrake proxy
        )

        # Mark as processed only after Stripe confirms.
        cur.execute("""
            UPDATE billing_records
            SET processed = TRUE,
                stripe_charge_id = %s,
                processed_at = NOW()
            WHERE customer_id = %s AND billing_period = %s
        """, (charge['id'], customer_id, billing_period))
        conn.commit()

    except stripe.error.CardError:
        # Card declined: mark as processed (declined) so retries skip it.
        cur.execute("""
            UPDATE billing_records
            SET processed = TRUE, status = 'card_declined', processed_at = NOW()
            WHERE customer_id = %s AND billing_period = %s
        """, (customer_id, billing_period))
        conn.commit()

    except stripe.error.Timeout:
        # Timeout: do NOT mark as processed.
        # Firehose will retry; the content-hash idempotency key returns
        # the original charge from Stripe if the charge was created
        # server-side. The pre-flight row exists with processed=False,
        # blocking the retry from entering the Stripe call path via the
        # ON CONFLICT check above — which is correct: the idempotency key
        # will resolve the outcome on the next attempt.
        raise

The UNIQUE constraint on (customer_id, billing_period) in PostgreSQL is the authoritative deduplication gate. A Firehose S3 retry that delivers the same billing records to a new Lambda invocation finds the pre-flight row and skips the Stripe call entirely. A Redshift COPY retry that inserts duplicate rows into billing_events still hits the PostgreSQL pre-flight check before calling Stripe — the duplicate Redshift row does not affect the pre-flight outcome. An HTTP endpoint retry with a different timestamp that produces a different SHA-256 idempotency key (if the developer mistakenly included the timestamp) is caught by the pre-flight row, which uses customer_id and billing_period as keys independent of Firehose metadata.

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

The pre-flight check and content-hash idempotency key close all failure modes within normal operating parameters. A per-billing-period vault key via a spend-cap proxy provides a hard backstop for the edge cases they cannot reach: a Firehose retry arriving after PostgreSQL has been restored from backup (pre-flight table rolled back), a disaster recovery scenario where the billing_records table is partially restored, or a concurrent billing job instance that pre-dates the UNIQUE constraint migration. The vault key caps total Stripe spend for a given customer and billing period at expected_total × 1.10:

# Issue one vault key per customer per billing period via Keybrake.
# Cap = expected charge amount × 1.10 (10% buffer for usage spikes).
# All workers processing any Firehose delivery for this customer+period
# share one vault key and one cap. Even if a duplicate charge somehow
# bypasses the pre-flight check and the Stripe idempotency window,
# the vault key cap prevents the total from exceeding 110% of expected.

import requests

def issue_vault_key(customer_id: str, billing_period: str,
                    expected_amount_cents: int) -> str:
    resp = requests.post('https://proxy.keybrake.com/vault/keys', json={
        'vendor': 'stripe',
        'label': f'{customer_id}:{billing_period}',
        'daily_usd_cap': round(expected_amount_cents * 1.10 / 100, 2),
        'allowed_endpoints': ['/v1/charges', '/v1/payment_intents'],
        'expires_at': f'{billing_period}-31T23:59:59Z',
    }, headers={'Authorization': f'Bearer {KEYBRAKE_API_KEY}'})
    return resp.json()['vault_key']

Protection comparison: five levels across three failure modes

Four additional failure surfaces to watch for in Firehose billing pipelines

Dynamic partitioning increases retry surface area. With dynamic partitioning enabled, Firehose extracts partition keys from record fields (e.g., customer_id) and routes records to customer-specific S3 prefixes like billing/customer_id=cus_abc123/year=2026/month=07/. Each partition is flushed independently when it hits its buffer threshold. A transient S3 error for one partition causes only that partition’s delivery to retry — but the retry creates a new UUID-suffixed S3 object in the same partition prefix. A billing Lambda subscribed to s3:ObjectCreated:Put on the partition prefix fires for both objects. The customer-specific partition prefix makes it tempting to use the partition path components (customer_id extracted from the prefix, year, month) as idempotency key components — but the UUID suffix still changes on retry. Use record-body fields only.

Lambda transformation timeout creates a partial-batch retry. Firehose optionally passes buffered batches through a Lambda transformation function before delivery to the destination. If the transformation Lambda times out, Firehose retries the transformation with the same batch. If the Lambda was partially through processing — some records transformed, some not — the retry runs the transformation again on the entire batch. The downstream delivery now contains records transformed twice. A billing enrichment Lambda that increments a usage counter or writes an audit entry for each record it transforms produces duplicate audit entries; if the enriched record body includes a transformation timestamp, the content hash of the record changes between the original and the retry, potentially breaking content-hash idempotency keys that hash the full record body rather than stable business fields only.

Firehose-to-S3 with S3 event notification to SQS: Lambda concurrency creates two concurrent billing jobs for the same object. If the SQS queue receives two notifications for the same S3 object (original + retry delivery, two different objects), and both messages become visible at the same time, two Lambda invocations run concurrently. The pre-flight PostgreSQL ON CONFLICT DO NOTHING handles this: both invocations attempt to insert the same (customer_id, billing_period) row; the UNIQUE constraint ensures only one INSERT succeeds; the other invocation’s INSERT returns 0 rows affected and the invocation skips the Stripe call. The concurrent case is handled correctly without additional locking.

Firehose error output prefix sends processing errors to a separate S3 prefix; reprocessing the error prefix without deduplication creates charges for events that were already billed. Firehose writes records that fail Lambda transformation or destination delivery to an error output prefix: billing/error/processing-failed/YYYY/MM/DD/HH/. Operators often build reprocessing pipelines that read error objects and re-run the billing logic. If the original delivery succeeded for some records before the failure, those records are also present in the error object. Reprocessing without the pre-flight PostgreSQL check or without a content-hash idempotency key creates duplicate charges for every already-billed event in the error object.

Five frequently asked questions

Does Firehose’s at-least-once guarantee mean duplicates are rare or common? Duplicates are rare in practice — AWS reports that S3 delivery duplicates occur when the first PutObject succeeds but the acknowledgment is lost, which is an infrastructure-level transient. But “rare” in a high-throughput billing pipeline that flushes every 5 minutes, 24/7, means one duplicate event every few weeks in a production system. The cost is not the frequency; it is the impact. A single duplicate in a billing pipeline creates an incorrect charge that must be detected, refunded, and explained to the customer. The pre-flight check costs one PostgreSQL INSERT per billing event; that is the right trade-off.

Can I use Firehose’s requestId as the Stripe idempotency key for HTTP endpoint destination? No, and for a specific reason: requestId is a batch-level identifier, not a record-level one. A Firehose batch can contain multiple billing records for different customers. Using requestId alone as the idempotency key would mean all records in the batch share one idempotency key — Stripe would treat the second stripe.charges.create() call with the same key as a duplicate of the first, regardless of the customer. You would charge only the first customer in the batch and silently skip all others. Use requestId for batch-level dedup (have you processed this batch before?) but derive per-record Stripe idempotency keys from customer_id and billing_period only.

Does a UNIQUE constraint on the Redshift billing_events table prevent duplicate rows from the COPY retry? Redshift does not enforce UNIQUE constraints — it accepts them as syntax for query optimizer hints but does not perform uniqueness checks at INSERT time. A Redshift COPY command inserts all rows from the staging file even if (customer_id, billing_period) already exists. The UNIQUE constraint that prevents duplicate billing is the one on the pre-flight PostgreSQL billing_records table, not on the Redshift billing_events table. If you use Redshift as your sole data store (no separate PostgreSQL), use a Redshift dedup CTE (ROW_NUMBER() OVER (PARTITION BY customer_id, billing_period ORDER BY inserted_at)) to extract one canonical row per (customer_id, billing_period) before calling Stripe, and apply the content-hash idempotency key to that canonical row.

Does S3 object versioning prevent Firehose retry duplicates? No. S3 object versioning means that if the same key is written twice, both versions are preserved. Firehose uses a different UUID suffix for the retry, so the original and retry objects have entirely different keys — versioning does not apply. Versioning would only help if Firehose retried with the same key, which it does not. Both objects exist as distinct keys, both fire distinct S3 event notifications, and both trigger independent Lambda invocations. The pre-flight check is the correct guard; versioning is orthogonal.

If the billing endpoint returns HTTP 200 on the first delivery, can Firehose still retry? Only in edge cases where the HTTP 200 response is received by Firehose but the response body format does not match Firehose’s expected format. Firehose HTTP endpoint requires the response body to include the requestId field from the request: {"requestId": "..."}. If the billing endpoint returns HTTP 200 with an empty body or a different JSON structure, Firehose may treat the delivery as failed (the behavior depends on Firehose version and endpoint configuration). Ensure your endpoint always returns jsonify({"requestId": body["requestId"]}), 200. The pre-flight check handles the pathological case where Firehose retries after a correctly-formatted HTTP 200, though this should not happen in practice.