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

Amazon Kinesis Data Streams is AWS’s managed real-time data streaming service. When a billing pipeline reads usage events from a Kinesis shard and calls stripe.charges.create() for each one, three Kinesis-specific behaviors — KCL shard lease coordination via DynamoDB, Enhanced Fan-Out session limits, and PutRecords partial-failure retry semantics — introduce Stripe double-charge failure modes that visibility timeout tuning and retry backoff alone cannot prevent.

This post covers those three failure modes with boto3 Python code, content-hash idempotency keys that stay stable across shard lease transfers and EFO re-subscriptions, 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 shard count, checkpoint frequency, or producer retry strategy.

Failure mode 1: KCL shard lease expiry during stripe.charges.create() — DynamoDB lease renewal fails under ProvisionedThroughputExceededException; competing worker acquires the expired lease and re-processes billing events from the last checkpoint; both workers call stripe.charges.create() for the same customers

The Kinesis Client Library (KCL) coordinates distributed shard ownership across consumer workers using a DynamoDB table (one row per shard) that stores the current lease owner, a lease counter, and the last checkpointed sequence number. Each KCL worker must renew its leases periodically — every failoverTimeMillis milliseconds, configurable via the KCL properties file (default 10 seconds in KCL 2.x). Lease renewal is a DynamoDB UpdateItem call that increments the leaseCounter attribute for each shard the worker owns. The write uses a conditional expression (attribute_exists(leaseCounter) AND leaseCounter = :current) to ensure only the current owner can renew.

During a billing run, all KCL workers simultaneously process events and checkpoint. Each checkpoint write is another UpdateItem call on the same DynamoDB table. With many shards, the DynamoDB checkpoint table receives concurrent writes from every shard worker at once: checkpoint updates after processing, plus lease renewals on the renewal schedule. If the table’s provisioned write capacity units are exhausted — or if the table is using on-demand capacity that is momentarily throttled — DynamoDB returns ProvisionedThroughputExceededException on the lease renewal call. The DynamoDB client retries with exponential backoff, but if retries are also throttled (throttling is self-reinforcing: backoff adds latency, during which the lease continues aging), the renewal may not complete within failoverTimeMillis.

When a lease reaches its expiry age, another KCL worker’s lease-balancing loop acquires it via a conditional UpdateItem (setting a new owner and incrementing the counter). The new owner reads the checkpointSequenceNumber from the DynamoDB row — the last position successfully committed — and begins reading Kinesis records from that position using get_shard_iterator with ShardIteratorType=AFTER_SEQUENCE_NUMBER.

The original worker is still inside stripe.charges.create() for billing events it received after the last successful checkpoint. The two workers now independently process the same events:

import boto3
import stripe
import hashlib
import json
import time

kinesis = boto3.client('kinesis', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
checkpoint_table = dynamodb.Table('kinesis-billing-checkpoints')

def process_shard(stream_name, shard_id):
    # Read current checkpoint from DynamoDB
    resp = checkpoint_table.get_item(Key={'shard_id': shard_id})
    last_seq = resp.get('Item', {}).get('checkpoint_sequence_number')

    if last_seq:
        iterator_resp = kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='AFTER_SEQUENCE_NUMBER',
            StartingSequenceNumber=last_seq
        )
    else:
        iterator_resp = kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='TRIM_HORIZON'
        )

    shard_iterator = iterator_resp['ShardIterator']

    while shard_iterator:
        resp = kinesis.get_records(ShardIterator=shard_iterator, Limit=100)

        for record in resp['Records']:
            event = json.loads(record['Data'])
            customer_id = event['customer_id']
            billing_period = event['billing_period']
            sequence_number = record['SequenceNumber']

            # UNSAFE: no idempotency key.
            # If DynamoDB lease renewal fails and another worker acquires
            # the lease, it reads from the last checkpoint and re-processes
            # all records since then. Both workers call stripe.charges.create()
            # for the same customers, creating ch_A (original worker) and
            # ch_B (new lease owner).
            charge = stripe.Charge.create(
                amount=event['amount_cents'],
                currency='usd',
                customer=event['stripe_customer_id'],
                description=f'Usage billing for {billing_period}',
                # missing: idempotency_key
            )

            # Checkpoint updated per-batch, not per-record.
            # DynamoDB write can fail under throughput pressure,
            # leaving last_seq behind the records just processed.
            checkpoint_table.update_item(
                Key={'shard_id': shard_id},
                UpdateExpression='SET checkpoint_sequence_number = :seq',
                ConditionExpression='attribute_not_exists(lease_owner) OR lease_owner = :me',
                ExpressionAttributeValues={
                    ':seq': sequence_number,
                    ':me': 'worker-1'
                }
            )

        shard_iterator = resp.get('NextShardIterator')
        time.sleep(1)

The lease expiry failure mode is compounded by two factors. First, it is correlated with billing load: ProvisionedThroughputExceededException is most likely during high-throughput billing runs precisely when all workers are writing checkpoints simultaneously. Second, the original worker’s stripe.charges.create() call typically takes 10–20 seconds under load — a window wider than failoverTimeMillis — and the worker has no mechanism to detect that it has lost the lease while the HTTP call is in-flight.

The new lease owner starts from the last checkpointed sequence number, which predates all records the original worker processed since its last checkpoint. With a checkpoint interval of 60 seconds and a stripe.charges.create() call at position 50 seconds into the interval, the new owner re-processes up to 50 seconds of billing events — potentially hundreds of duplicate charges.

Failure mode 2: Enhanced Fan-Out SubscribeToShard session limit and crash-between-stripe-and-checkpoint — re-subscription from the last DynamoDB checkpoint re-delivers records processed since the last checkpoint, calling stripe.charges.create() for already-billed customers

Enhanced Fan-Out (EFO) gives each registered consumer a dedicated 2 MB/s per shard throughput via SubscribeToShard — records are pushed via HTTP/2 server-sent events instead of polled with GetRecords. EFO eliminates read-throttling from multiple consumers sharing a shard’s 2 MB/s limit, which makes it popular for high-throughput billing pipelines where polling consumers would contend.

Each SubscribeToShard call establishes a session that lasts at most 5 minutes. After 5 minutes, the HTTP/2 stream is closed by AWS and the consumer must call SubscribeToShard again. The re-subscription typically uses StartingPosition=AFTER_SEQUENCE_NUMBER pointing to the last sequence number the consumer received from the previous session. The failure scenario arises in two related ways.

Scenario A: A billing callback processes records as they arrive from the SubscribeToShard event stream, calling stripe.charges.create() for each and checkpointing to DynamoDB every N records (or every T seconds). The consumer crashes — OOM kill, EC2 instance termination, SIGKILL from container orchestrator — between calling stripe.charges.create() and writing the checkpoint to DynamoDB. On restart, the consumer reads the last checkpointed sequence number from DynamoDB and re-subscribes from that position. Every record processed since the last checkpoint is re-delivered to the billing callback.

Scenario B: A consumer re-subscribes with StartingPosition=AT_TIMESTAMP as a fallback after a connectivity error where the last in-memory sequence number was lost. If the timestamp used for re-subscription overlaps with records from a prior billing run — a common mistake in time-based recovery implementations — records already processed and billed are re-delivered.

import boto3
import stripe
import hashlib
import json
import time

kinesis = boto3.client('kinesis', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
checkpoint_table = dynamodb.Table('kinesis-billing-checkpoints')

CONSUMER_ARN = 'arn:aws:kinesis:us-east-1:123456789012:stream/billing/consumer/billing-efo-consumer/...'

def subscribe_and_process(shard_id):
    resp = checkpoint_table.get_item(Key={'shard_id': shard_id})
    last_seq = resp.get('Item', {}).get('checkpoint_sequence_number')

    starting_position = (
        {'Type': 'AFTER_SEQUENCE_NUMBER', 'SequenceNumber': last_seq}
        if last_seq
        else {'Type': 'TRIM_HORIZON'}
    )

    response = kinesis.subscribe_to_shard(
        ConsumerARN=CONSUMER_ARN,
        ShardId=shard_id,
        StartingPosition=starting_position
    )

    processed_since_checkpoint = 0
    last_processed_seq = last_seq

    for event in response['EventStream']:
        if 'SubscribeToShardEvent' not in event:
            continue

        for record in event['SubscribeToShardEvent']['Records']:
            billing_event = json.loads(record['Data'])
            customer_id = billing_event['customer_id']
            billing_period = billing_event['billing_period']
            seq = record['SequenceNumber']

            # UNSAFE: idempotency key derived from sequence_number.
            # Sequence numbers are assigned at ingestion time and change
            # for every PutRecords retry copy. On re-subscription after
            # crash-between-stripe-and-checkpoint, the same record arrives
            # with the SAME sequence_number — but if this is a PutRecords
            # retry copy, it has a DIFFERENT sequence_number, producing a
            # different idempotency key. Stripe creates both charges.
            idempotency_key = hashlib.sha256(
                f'{customer_id}:{billing_period}:{seq}'.encode()
            ).hexdigest()[:32]

            charge = stripe.Charge.create(
                amount=billing_event['amount_cents'],
                currency='usd',
                customer=billing_event['stripe_customer_id'],
                description=f'Usage billing for {billing_period}',
                idempotency_key=idempotency_key,  # broken: seq changes on retry copies
            )

            last_processed_seq = seq
            processed_since_checkpoint += 1

            # Checkpoint every 100 records.
            # If consumer crashes between stripe call above and checkpoint
            # below, re-subscription from DynamoDB restarts BEFORE this
            # record — stripe.charges.create() is called again on restart.
            # The sequence_number-based idempotency key matches for the
            # re-delivered original record (same seq), but NOT for a
            # PutRecords retry copy (different seq on the retry copy).
            if processed_since_checkpoint >= 100:
                checkpoint_table.put_item(Item={
                    'shard_id': shard_id,
                    'checkpoint_sequence_number': last_processed_seq,
                })
                processed_since_checkpoint = 0

    # SubscribeToShard session expired (5-minute limit).
    # Re-subscribe from last_processed_seq.
    # If a crash occurred above, we re-subscribe from the DynamoDB checkpoint.
    subscribe_and_process(shard_id)  # recursive re-subscription

The crash-between-stripe-and-checkpoint window is present in every consume-and-checkpoint loop, regardless of whether the consumer uses polling (GetRecords) or Enhanced Fan-Out (SubscribeToShard). The EFO-specific addition is the mandatory 5-minute re-subscription: even a consumer that never crashes must re-establish its session, and any re-subscription implementation that does not correctly track the in-memory last-processed sequence number across the session boundary risks re-delivering records.

Failure mode 3: PutRecords partial failure — producer retries all records including successfully written ones; Kinesis assigns new sequence numbers to retry copies; downstream consumer’s sequence_number-based idempotency key produces a different key for each copy; Stripe creates both charges

Kinesis PutRecords accepts up to 500 records per call and returns per-record results in the response. A record is either accepted (the response includes a SequenceNumber and ShardId) or rejected with an ErrorCode (ProvisionedThroughputExceededException or InternalFailure) and an ErrorMessage. The response’s FailedRecordCount field indicates how many records failed in the batch.

AWS documentation and best practices recommend retrying only the failed records, correlating them by position index in the original request array. However, many production implementations retry the entire batch when FailedRecordCount > 0, for one of three reasons: the client library abstracts PutRecords into a simple put(record) interface that doesn’t expose per-record retry selection; the application uses botocore’s built-in retry configuration which retries the entire PutRecords call on retryable errors; or the engineering team adopted a simpler “retry the whole batch” strategy to avoid correlation bookkeeping bugs.

When the entire batch is retried, the successfully written records are written again. Kinesis assigns new sequence numbers to the retry copies — sequence numbers are monotonically increasing integers assigned at ingestion time and are unique across the entire stream. The two copies of the same billing event exist independently in the shard:

import boto3
import stripe
import hashlib
import json

kinesis = boto3.client('kinesis', region_name='us-east-1')

def publish_billing_events(stream_name, events):
    records = [
        {
            'Data': json.dumps(event).encode('utf-8'),
            'PartitionKey': event['customer_id'],
        }
        for event in events
    ]

    response = kinesis.put_records(
        StreamName=stream_name,
        Records=records,
    )

    # UNSAFE: retry the ENTIRE batch when any record fails.
    # Records that succeeded in the first call are re-written with
    # new sequence numbers. The consumer receives two copies:
    #   copy 1: SequenceNumber=49569374293798716849250261
    #   copy 2: SequenceNumber=49569374293798716849250277
    # A consumer using sha256(customer_id:billing_period:SequenceNumber)
    # produces different idempotency keys for each copy.
    # Stripe creates ch_A for copy 1 and ch_B for copy 2.
    if response['FailedRecordCount'] > 0:
        publish_billing_events(stream_name, events)  # retries ALL events

# ----- Consumer side: sequence_number in the idempotency key -----
def process_record_unsafe(record):
    event = json.loads(record['Data'])
    seq = record['SequenceNumber']  # 49569374293798716849250261 (original)
                                    # 49569374293798716849250277 (retry copy)

    idempotency_key = hashlib.sha256(
        f'{event["customer_id"]}:{event["billing_period"]}:{seq}'.encode()
    ).hexdigest()[:32]
    # Key for copy 1: sha256('cus_X:2026-07:49569...261')[:32] = 'a1b2c3...'
    # Key for copy 2: sha256('cus_X:2026-07:49569...277')[:32] = 'd4e5f6...'
    # Stripe sees two different idempotency keys — two separate charge requests.

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

The PutRecords retry failure also surfaces in indirect paths. An AWS Lambda function with a Kinesis event source mapping that throws an exception causes Lambda to retry the batch from the shard position of the failing batch start. If the Lambda function called PutRecords to forward events to a second Kinesis stream (a fan-out pattern), and that forwarding call used a full-batch retry strategy, duplicate records exist in the second stream even though the Lambda Kinesis event source delivers the first stream’s records exactly once. A Kinesis Data Firehose destination with buffering enabled can also re-deliver buffered records if the flush fails, writing duplicates to S3 or Redshift that a downstream billing consumer then double-processes.

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

All three failure modes share a common root cause: the idempotency key or deduplication mechanism uses a Kinesis-layer identifier (sequence number, shard ID, or iterator position) that changes across shard lease transfers, EFO re-subscriptions, or PutRecords retry copies. The fix is to derive the idempotency key exclusively from billing-domain fields that are stable across all Kinesis-layer operations.

The correct idempotency key for Kinesis billing:

import hashlib

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Must NOT include:
    #   - SequenceNumber: changes for every PutRecords retry copy
    #   - ShardId: different copy may route to different shard on retry
    #   - arrival_timestamp / ApproximateArrivalTimestamp: set at ingestion;
    #     later for retry copies
    #   - PartitionKey: same value for original and retry, but a content-
    #     addressable hash already encodes the business key
    #   - Any KCL lease metadata: not available in the record body
    #
    # Must include ONLY stable business fields:
    return hashlib.sha256(
        f'{customer_id}:{billing_period}:kinesis-billing'.encode()
    ).hexdigest()[:32]

The full pattern combines the idempotency key with a pre-flight database check against PostgreSQL and a write-before-checkpoint ordering:

import boto3
import stripe
import hashlib
import psycopg2
import json
import time
import logging

logger = logging.getLogger(__name__)

kinesis = boto3.client('kinesis', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
checkpoint_table = dynamodb.Table('kinesis-billing-checkpoints')

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

def process_billing_record(record, pg_conn):
    event = json.loads(record['Data'])
    customer_id = event['customer_id']
    billing_period = event['billing_period']
    amount_cents = event['amount_cents']
    stripe_customer_id = event['stripe_customer_id']

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    # Pre-flight check against PostgreSQL — authoritative dedup guard.
    # PostgreSQL WAL is fsynced on commit and survives:
    #   - KCL shard lease expiry and lease transfer (DynamoDB-layer event)
    #   - EFO SubscribeToShard session expiry and re-subscription
    #   - Consumer crash-between-stripe-and-checkpoint
    #   - PutRecords retry copies (same content-hash key, same row)
    with pg_conn.cursor() as cur:
        cur.execute(
            'SELECT charge_id, status FROM billing_records '
            'WHERE customer_id = %s AND billing_period = %s',
            (customer_id, billing_period)
        )
        row = cur.fetchone()

    if row is not None:
        existing_charge_id, existing_status = row
        logger.info(
            'Pre-flight check: already have %s for %s/%s, skipping Stripe call',
            existing_status, customer_id, billing_period
        )
        return existing_charge_id

    # Issue vault key (Keybrake) scoped to this billing period.
    # Vault key capped at expected_amount_cents * 1.10 — hard backstop
    # for PutRecords retry duplicates processed after Stripe's 24-hour
    # idempotency window closes, or parallel lease-steal re-processing.
    vault_key = get_vault_key_for_period(
        customer_id=customer_id,
        billing_period=billing_period,
        cap_cents=int(amount_cents * 1.10),
    )

    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,
        )

    except stripe.error.CardError:
        # Permanent failure — write to billing_records with status='card_declined'.
        # Prevents KCL re-delivery or EFO re-subscription from repeatedly
        # firing stripe.charges.create() for an unchargeable customer.
        _write_billing_record(pg_conn, customer_id, billing_period,
                              None, amount_cents, 'card_declined')
        return None

    except stripe.error.Timeout:
        # ch_A may already exist on Stripe's servers.
        # Do NOT write to billing_records — let the record be re-delivered.
        # The content-hash idempotency key will return ch_A on re-delivery.
        # Do NOT checkpoint this sequence number.
        raise

    # Write to PostgreSQL BEFORE updating the Kinesis checkpoint.
    # Write-before-checkpoint: if the JVM is killed between this write and
    # the DynamoDB checkpoint update, the PostgreSQL row survives.
    # The re-delivered record's pre-flight check finds the row and skips
    # stripe.charges.create() — regardless of whether the re-delivery comes
    # from a KCL lease transfer, an EFO re-subscription, or a PutRecords
    # retry copy (same content-hash key = same pre-flight row found).
    # ON CONFLICT DO NOTHING closes the concurrent-processing race where
    # both the original worker and the new lease owner simultaneously
    # pass the pre-flight check and both attempt the INSERT.
    _write_billing_record(pg_conn, customer_id, billing_period,
                          charge.id, amount_cents, 'charged')

    return charge.id

def _write_billing_record(pg_conn, customer_id, billing_period,
                          charge_id, amount_cents, status):
    with pg_conn.cursor() as cur:
        cur.execute(
            '''
            INSERT INTO billing_records
              (customer_id, billing_period, charge_id, amount_cents, status, created_at)
            VALUES (%s, %s, %s, %s, %s, NOW())
            ON CONFLICT (customer_id, billing_period) DO NOTHING
            ''',
            (customer_id, billing_period, charge_id, amount_cents, status)
        )
    pg_conn.commit()

def process_shard_safe(stream_name, shard_id, pg_conn):
    resp = checkpoint_table.get_item(Key={'shard_id': shard_id})
    last_seq = resp.get('Item', {}).get('checkpoint_sequence_number')

    if last_seq:
        it = kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='AFTER_SEQUENCE_NUMBER',
            StartingSequenceNumber=last_seq,
        )['ShardIterator']
    else:
        it = kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='TRIM_HORIZON',
        )['ShardIterator']

    batch_last_seq = None

    while it:
        resp = kinesis.get_records(ShardIterator=it, Limit=10)

        for record in resp['Records']:
            # process_billing_record writes to PostgreSQL before returning.
            # Any re-delivery (lease steal, EFO re-subscribe, retry copy)
            # hits the pre-flight check and skips the Stripe call.
            process_billing_record(record, pg_conn)
            batch_last_seq = record['SequenceNumber']

        if batch_last_seq:
            # Checkpoint to DynamoDB AFTER PostgreSQL write completes.
            # If this checkpoint write fails (ProvisionedThroughputExceededException),
            # the lease may be stolen. But the PostgreSQL billing_records row
            # is already written — the new owner's pre-flight check skips Stripe.
            try:
                checkpoint_table.update_item(
                    Key={'shard_id': shard_id},
                    UpdateExpression='SET checkpoint_sequence_number = :seq',
                    ExpressionAttributeValues={':seq': batch_last_seq},
                )
            except Exception as e:
                logger.warning('Checkpoint write failed: %s — billing_records is durable', e)
                # Do NOT raise — PostgreSQL has the billing row.
                # If the lease is stolen, new owner's pre-flight check handles it.

        it = resp.get('NextShardIterator')
        if not resp['Records']:
            time.sleep(1)

def get_vault_key_for_period(customer_id, billing_period, cap_cents):
    # Keybrake API: issue or retrieve vault key for this customer+period.
    # In practice, store the vault key in billing_records or a separate table
    # so that re-deliveries use the same vault key (and spend cap is not reset).
    pass  # implementation omitted

The safe PutRecords producer retries only failed records by index:

import boto3
import json
import time

kinesis = boto3.client('kinesis', region_name='us-east-1')

def publish_billing_events_safe(stream_name, events):
    records = [
        {
            'Data': json.dumps(event).encode('utf-8'),
            'PartitionKey': event['customer_id'],
        }
        for event in events
    ]

    while records:
        response = kinesis.put_records(
            StreamName=stream_name,
            Records=records,
        )

        if response['FailedRecordCount'] == 0:
            break

        # Retry ONLY the failed records.
        # Successfully written records already have SequenceNumbers in the stream.
        # Retrying them would create duplicate records with new sequence numbers.
        failed_records = []
        for i, result in enumerate(response['Records']):
            if 'ErrorCode' in result:
                failed_records.append(records[i])

        records = failed_records

        if records:
            time.sleep(0.5)  # back off before retry

Comparison: protection levels across all three failure modes

Gap analysis

Kinesis Firehose buffering delivers batched records to S3 or Redshift; a Firehose delivery retry after an S3 write timeout can write the same records twice to the destination; a billing pipeline reading from S3 or Redshift processes both copies

Kinesis Data Firehose buffers incoming records and delivers them in batches (buffered by size up to 128 MB or time up to 15 minutes). If a delivery attempt to S3 fails (S3 returns a 5xx error or the delivery times out), Firehose retries the entire buffer. On success, the same records are written to S3 twice under different keys (Firehose includes the retry attempt in the object key). A billing pipeline that reads from S3 — for example, a Lambda triggered by S3 PutObject events, or an AWS Glue job over the S3 prefix — processes both copies of the billing records. The content-hash idempotency key handles this: both copies produce the same idempotency key for each customer+period pair, and the pre-flight PostgreSQL check skips the Stripe call for the second copy. Ensure the S3 billing reader uses the same deduplication pattern as the direct Kinesis consumer.

Kinesis shard split and merge change the shard topology; a billing consumer that derives any part of its dedup key from the shard ID processes records from parent shards and child shards with different shard IDs

Kinesis shard splits create two child shards from one parent. After a split, the parent shard is sealed (no new records) and existing records are read from it, then new records arrive in the child shards. A KCL consumer will process records from the parent shard and then from the two child shards. If a billing consumer derives any part of its idempotency key or dedup check from the ShardId, records that were in-flight during the split boundary — checkpointed in the parent shard but not yet re-read from the child shards — could produce a different key when re-processed by the child shard readers. The content-hash idempotency key derived from customer_id and billing_period only is shard-agnostic: it produces the same key regardless of which shard the record came from, making the dedup stable across all shard topology changes.

Kinesis iterator expiration: GetRecords returns an empty response if the iterator is more than 5 minutes old; a polling consumer that sleeps between polls may receive an expired-iterator error and re-request the iterator from the last checkpoint, re-reading already-processed records

A shard iterator returned by get_shard_iterator or as NextShardIterator in a get_records response is valid for up to 5 minutes. A polling consumer that sleeps for more than 5 minutes between get_records calls (for example, a low-throughput billing pipeline that sleeps when there are no records) will receive an ExpiredIteratorException on the next call. If the error-recovery code requests a new iterator using AFTER_SEQUENCE_NUMBER pointing to the last checkpointed sequence number (not the last polled sequence number), records between the last checkpoint and the sleep period are re-read. The pre-flight PostgreSQL check is the guard here: the re-read records’ pre-flight check finds billing_records rows for all customers billed since the last checkpoint and skips the Stripe call.

Kinesis on-demand scaling adjusts shard count automatically; during a scale-out event, some records may be temporarily throttled and a botocore retry for the entire PutRecords call re-sends already-accepted records

Kinesis on-demand mode auto-scales shard count up to double the prior peak within 30 minutes. During the scale-out ramp, throughput limits are temporarily below the configured target, and individual PutRecords records may receive ProvisionedThroughputExceededException. If the application’s botocore retry configuration retries the entire PutRecords call (not just the failed records), already-accepted records are re-sent. The safe producer that retries only failed records by index (shown above) prevents this. The content-hash idempotency key on the consumer side provides a backstop if retry copies reach the consumer despite correct producer-side retry logic.

FAQ

Can I use Kinesis’s AFTER_SEQUENCE_NUMBER shard iterator type as a dedup mechanism by always starting exactly one record past the last processed?

No. AFTER_SEQUENCE_NUMBER tells Kinesis where to start reading — it prevents re-reading records before the specified sequence number. It does not prevent duplicate processing of records after that sequence number. In the KCL lease-steal failure mode, the competing worker starts from the checkpointed sequence number (not the last processed one), so it re-reads and re-processes all records between the checkpoint and the crash. AFTER_SEQUENCE_NUMBER starts the new owner at exactly the right checkpoint position — which is the position that causes the re-delivery. The pre-flight PostgreSQL check is the mechanism that detects the re-delivery and skips the Stripe call.

Does Kinesis Enhanced Fan-Out’s exactly_once delivery guarantee prevent duplicate billing?

Enhanced Fan-Out does not offer an exactly-once delivery guarantee. Kinesis documentation describes EFO as providing dedicated throughput per consumer, not exactly-once semantics. Records can be re-delivered on EFO re-subscription (after the 5-minute session limit, after a network interruption, or after a consumer restart from a DynamoDB checkpoint). The content-hash idempotency key and pre-flight PostgreSQL check are required for exactly-once billing regardless of whether the consumer uses standard polling or Enhanced Fan-Out.

How should I set the Kinesis checkpoint frequency to minimize duplicate charge exposure?

Checkpoint frequency controls how many records are re-delivered after a lease steal or consumer restart, but it is a tuning knob, not a correctness guarantee. Checkpointing after every record (per-record checkpointing) minimizes re-delivery to at most one record per failure, but significantly increases DynamoDB write throughput pressure — which is also the trigger for the KCL lease renewal failure that causes the duplicate. Checkpointing every 10–100 records is a common balance. Regardless of checkpoint frequency, the pre-flight PostgreSQL check is the authoritative dedup guard: even with per-record checkpointing, the write-before-checkpoint ordering ensures the billing_records row is durable before the checkpoint is updated, and any re-delivery is caught by the pre-flight check.

How should I issue the Keybrake vault key for a Kinesis billing pipeline that processes events from multiple shards in parallel?

Issue one vault key per customer per billing period, not one per shard. Each Kinesis shard is processed by an independent KCL worker, but multiple shards may contain events for the same customer (if events are published without a customer-specific partition key, or if a customer’s events span a shard split boundary). A vault key scoped to the customer and billing period with a cap of expected_amount_cents * 1.10 is enforced regardless of which shard the billing event arrives from. Store the vault key in the billing_records table (or a dedicated vault_keys table) so that all shard workers for the same customer+period use the same vault key — and the spend cap is shared across all parallel processing paths. A separate vault key per shard would multiply the cap by the shard count, making the cap meaningless as a governance control.

Does the ON CONFLICT DO NOTHING on the PostgreSQL billing_records INSERT mean the second concurrent INSERT silently succeeds without creating a charge?

Yes, and this is the correct behavior. ON CONFLICT DO NOTHING means that if two concurrent processors (e.g., the original KCL worker and the new lease owner) both pass the pre-flight check simultaneously and both attempt to INSERT the same (customer_id, billing_period) row, only one INSERT wins the unique constraint race. The other INSERT silently inserts zero rows. The winning processor’s stripe.charges.create() call creates the charge; the losing processor’s Stripe call also completes (it ran before the pre-flight check race was resolved), but the content-hash idempotency key ensures Stripe returns the same charge ID for both calls. The losing processor’s INSERT fails silently; the billing_records table has exactly one row for the customer+period. The ON CONFLICT DO NOTHING is not suppressing an error — it is implementing the desired idempotent-insert semantics that make concurrent processing safe.