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

Meltano is a Singer-based data integration platform that orchestrates taps (data sources) and targets (data destinations) in reproducible ELT pipelines. Teams wire Meltano into Stripe billing pipelines to trigger charges based on usage events extracted from product databases, SaaS APIs, and event stores via Singer taps: the tap emits usage records to a Singer target, a downstream billing function reads those records and calls stripe.charges.create() for each customer, and the result is committed back to a billing database. Three Meltano-specific behaviors introduce Stripe double-charge failure modes that Meltano’s job logs, Singer state backend, and pipeline health dashboards do not surface as billing risk.

This post covers those three failure modes with Meltano YAML configuration, Singer tap Python code, content-hash idempotency keys, per-run vault keys via a spend-cap proxy, and pre-flight database checks — the two-layer governance pattern that eliminates all three without modifying Meltano’s pipeline architecture.

Failure mode 1: Singer tap state saved only on clean pipeline exit — failed run re-emits from last bookmark on retry

Meltano’s incremental pipelines rely on the Singer state mechanism: the tap emits STATE messages throughout the run that record how far through the source dataset the extractor has read. Meltano collects these state messages and — on successful pipeline exit — writes the final state to its state backend (the Meltano system database or a configured external state store). The next meltano run reads this saved state and passes it to the tap, which resumes from the last bookmark.

The critical behavior: Meltano writes the Singer state only after the pipeline exits with code 0. If the tap emits 900 records to the target, the target writes 900 rows to the billing staging table, and then the target raises a connection error while writing row 901, Meltano terminates the pipeline run as failed. The state messages the tap emitted during that run are discarded — Meltano does not partially save state on failure. The next meltano run starts from the state that was saved at the end of the last successful run, which may be many records before the 900 rows already written to the staging table.

If a billing function reads from the staging table after each pipeline run and calls stripe.charges.create() for each new row, the retry delivers 900 rows that the billing function treats as new usage events — because each row carries a fresh extraction timestamp from the retry run. Deduplication logic based on _sdc_received_at (the Singer Data Collector timestamp written by most Singer targets) does not protect against this: _sdc_received_at is the time the target received the record, not a stable business key. It changes on every extraction run.

# meltano.yml — partial pipeline config
plugins:
  extractors:
    - name: tap-postgres
      variant: meltanolabs
      pip_url: git+https://github.com/MeltanoLabs/tap-postgres.git
      config:
        sqlalchemy_url: ${DATABASE_URL}
        stream_maps:
          usage_events:
            # Filter only events since last billing run via Singer state bookmark
            __filter__: record['billed'] == false

  loaders:
    - name: target-jsonl
      variant: andyh1203
      pip_url: target-jsonl

schedules:
  - name: billing-extract
    extractor: tap-postgres
    loader: target-jsonl
    interval: "@daily"

# UNSAFE: billing trigger that reads from Singer target output after each meltano run
# If the pipeline failed and retried, the same records are re-delivered to the target
# with new _sdc_received_at values, and this billing function charges them a second time.

import stripe, os, json, psycopg2, pathlib

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # UNSAFE: unrestricted, shared

def run_billing_after_meltano():
    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    cur = conn.cursor()

    output_file = pathlib.Path("/meltano/output/usage_events.jsonl")
    for line in output_file.read_text().splitlines():
        record = json.loads(line)
        customer_id = record["customer_id"]
        amount_cents = record["amount_cents"]
        billing_period = record["billing_period"]
        sdc_received_at = record["_sdc_received_at"]  # changes on every meltano run

        # DANGER: uses _sdc_received_at as dedup key -- new timestamp on each retry.
        # A failed-then-retried pipeline delivers the same business events with
        # new _sdc_received_at values. This NOT IN check passes for retried rows.
        cur.execute(
            "SELECT 1 FROM billed_events WHERE sdc_received_at = %s",
            (sdc_received_at,)
        )
        if cur.fetchone():
            continue

        charge = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Billing for {billing_period}",
            # no idempotency_key — a retry creates ch_B for the same billing event
        )
        cur.execute(
            "INSERT INTO billed_events (sdc_received_at, charge_id) VALUES (%s, %s)",
            (sdc_received_at, charge["id"])
        )
    conn.commit()

Meltano’s job log records the failed run as FAILED and the retry as a new RUNNING then SUCCEEDED job. The tap’s extracted record count in the Meltano UI increases by 900 for the retry run, in addition to whatever count the failed run recorded. This looks like a larger-than-expected extraction, not a billing problem. The billing staging table now has 1,800 rows for 900 unique business events — 900 from the failed run and 900 from the successful retry. A billing trigger that reads all rows since the last billing timestamp (using _sdc_received_at as the watermark) fires for all 1,800.

Failure mode 2: concurrent meltano run invocations read the same initial Singer state — both emit unbilled records and fire Stripe independently

Meltano does not enforce at-most-one-run-per-pipeline by default. A scheduled pipeline run (via meltano schedule run or a cron that calls meltano run) that takes longer than the schedule interval can overlap with the next scheduled invocation. Similarly, an on-call engineer who triggers a manual meltano run to replay a suspected data gap while a scheduled run is already executing creates two concurrent pipeline instances.

Both concurrent runs read the Singer state from Meltano’s state backend at startup. Both read the same state — the state written by the last successful run before either of the current two runs started. Both taps query the source with the same filter derived from that bookmark, for example WHERE usage_at > '2026-07-16T00:00:00Z'. Both emit the same unbilled usage records to their respective target instances. If the billing function is downstream of the Singer target (for example, a Lambda triggered by an S3 upload from the Singer target), two independent Lambda invocations fire for the same customer dataset. Both call stripe.charges.create() for each customer. Both calls succeed if they do not share an idempotency key — resulting in two charges per customer.

# Both concurrent runs read this state at startup:
# {"bookmarks": {"usage_events": {"replication_key_value": "2026-07-16T00:00:00Z"}}}

# Run A (scheduled): starts at 01:00, reads bookmark 2026-07-16, emits 400 events
# Run B (manual replay): starts at 01:03, reads SAME bookmark 2026-07-16, emits 400 events

# Both target outputs land in S3 as separate objects:
# s3://meltano-output/usage_events/2026-07-17T010000Z/part-00000.jsonl  (Run A)
# s3://meltano-output/usage_events/2026-07-17T010347Z/part-00000.jsonl  (Run B)

# Each S3 PUT triggers a separate Lambda invocation.
# Lambda A processes 400 customers, calls stripe.charges.create() for each.
# Lambda B processes the SAME 400 customers, calls stripe.charges.create() again.
# Stripe's idempotency window closes this if both calls use identical idempotency keys,
# but if the keys include any run-specific metadata (timestamp, filename, S3 key path),
# Stripe creates ch_B for each customer — 400 duplicate charges in under 5 minutes.

Meltano’s state backend records both runs independently. Whichever run finishes last writes its end-of-run state to the backend, overwriting the other run’s state. If Run A finishes before Run B, Run B’s state (which covers the same bookmark range) is the final saved state. The next scheduled run starts from that state — which is identical to where Run A started, so no records are skipped from the source’s perspective. But 800 Stripe charges have already been created for 400 customers: 400 from Run A and 400 from Run B.

Meltano’s job UI shows two SUCCEEDED pipeline runs. Both extracted 400 records and loaded them successfully. Nothing in Meltano’s output indicates that the same business events were processed twice.

Failure mode 3: meltano run --no-state-update replays full tap history — charges every historical customer

meltano run --no-state-update tells Meltano not to read the saved Singer state at startup and not to write a new state at the end of the run. The tap starts from the beginning of its incremental range — for most taps, the source’s earliest available record. This is the intended behavior for CI pipelines and development environments where you want repeatable, deterministic full-extract runs that are independent of whatever state the production pipeline has saved.

The problem: teams that develop and test their Meltano billing pipelines in CI frequently use --no-state-update (sometimes via meltano config meltano set run.state_update_exclusion_patterns '*' applied globally in meltano.yml) to avoid polluting the production state backend during tests. When that configuration is accidentally applied in a production deployment — or when a developer runs the production Meltano instance locally with --no-state-update but with the production .env file containing the real Stripe API key — the tap emits every usage record from the source’s origin.

For a product database with two years of usage history, --no-state-update causes the tap to emit every usage event since product launch. The downstream billing function has no state-based signal to distinguish “this is a full-history replay” from “this is a normal incremental batch” — it receives records and calls Stripe for each one. Every customer who has ever been billed receives another charge.

# The --no-state-update flag is easy to pass accidentally:

# In CI (intended):
meltano run tap-postgres target-jsonl --no-state-update

# Accidentally applied to the production run via meltano.yml:
# meltano.yml
default_environment: production
environments:
  - name: production
    config:
      plugins:
        extractors:
          - name: tap-postgres
            # State exclusions set here for CI are still active in production
            # if the meltano.yml is deployed without review
            # (no flag needed — --no-state-update behavior is the default
            # when state_update_exclusion_patterns matches all streams)

# Alternative: developer runs meltano locally against prod .env
# .env contains STRIPE_SECRET_KEY=sk_live_... and DATABASE_URL=postgres://prod-db/...
# Developer adds --no-state-update to "avoid messing up the state"
# Tap emits 3 years of usage events. Billing lambda charges 8,000 historical customers.

meltano run tap-postgres target-jsonl --no-state-update \
  --environment production  # uses production credentials from .env

Meltano’s job log records this run as SUCCEEDED with a higher-than-usual extracted record count. Without a state update, the production state backend is unchanged after the run. The next scheduled incremental run sees the last saved state and extracts only the records since the last production bookmark — but by then, the billing function has already charged every historical customer from the full-history replay.

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

All three failure modes share the same root cause: the downstream billing trigger fires on records it has already processed. The fix has two independent layers that each close all three modes without requiring changes to Meltano’s pipeline configuration.

Layer 1: content-hash idempotency key stable across Singer extractions

The Stripe idempotency key must be derived from the business event’s stable fields — fields that do not change across Meltano retries, concurrent runs, or --no-state-update replays. Singer target metadata fields are extraction-time artifacts: _sdc_received_at is when the target received the record; _sdc_sequence is an integer derived from extraction order; _sdc_batched_at is when the batch was written. All three change on every extraction run. Do not include them in the idempotency key.

The key must be derived from the business event’s identity fields: customer_id, amount_cents, billing_period. Append a namespace suffix to scope the key to this pipeline:

import hashlib, stripe, os, psycopg2, json

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

def make_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    # Must NOT include: _sdc_received_at, _sdc_sequence, _sdc_batched_at,
    # extraction timestamp, meltano run ID, S3 object key, or any other
    # Singer/Meltano runtime artifact. These change across retries, concurrent
    # runs, and --no-state-update replays — making the key unstable.
    raw = f"{customer_id}:{amount_cents}:{billing_period}:meltano-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def process_usage_record(conn, record: dict) -> dict:
    customer_id = record["customer_id"]
    stripe_customer_id = record["stripe_customer_id"]
    amount_cents = record["amount_cents"]
    billing_period = record["billing_period"]
    # _sdc_received_at and _sdc_sequence are NOT used in the key or dedup logic

    cur = conn.cursor()

    # Pre-flight: skip if already billed (survives replays beyond Stripe's 24h window)
    cur.execute(
        "SELECT 1 FROM billing_records WHERE customer_id = %s AND billing_period = %s",
        (customer_id, billing_period)
    )
    if cur.fetchone():
        return {"status": "skipped", "customer_id": customer_id}

    idempotency_key = make_idempotency_key(customer_id, amount_cents, billing_period)

    # Use vault key instead of raw Stripe key — enforces per-run spend cap
    response = 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(
        "INSERT INTO billing_records (customer_id, billing_period, charge_id, billed_at) "
        "VALUES (%s, %s, %s, NOW()) ON CONFLICT (customer_id, billing_period) DO NOTHING",
        (customer_id, billing_period, response["id"])
    )
    conn.commit()
    return {"status": "charged", "charge_id": response["id"]}


def run_billing_after_meltano(output_file: str):
    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    results = []
    for line in open(output_file):
        if not line.strip():
            continue
        record = json.loads(line)
        result = process_usage_record(conn, record)
        results.append(result)
    return results

Layer 2: per-run vault key capped at expected total × 1.10

Issue a vault key before each Meltano billing run capped at 110% of the maximum expected total for that billing period. The vault key is passed as the STRIPE_SECRET_KEY environment variable to the billing function — or set as stripe.api_key at the start of the run — instead of the raw Stripe restricted key. Any billing run that exceeds the cap (because a concurrent run already exhausted it, because a --no-state-update replay started billing historical customers, or because a data error inflated amount_cents) returns a 402 spend_cap_exceeded response from the proxy after the cap is hit. No further charges clear until a new vault key is issued for the next billing period.

import requests, os

PROXY_BASE = "https://proxy.keybrake.com"
PROXY_KEY = os.environ["KEYBRAKE_API_KEY"]

def issue_billing_vault_key(billing_period: str, expected_total_usd: float) -> str:
    """Issue a per-billing-period vault key capped at expected_total * 1.10."""
    cap = round(expected_total_usd * 1.10, 2)
    resp = requests.post(
        f"{PROXY_BASE}/vault/keys",
        headers={"Authorization": f"Bearer {PROXY_KEY}"},
        json={
            "label": f"meltano-billing-{billing_period}",
            "vendor": "stripe",
            "daily_usd_cap": cap,
            "allowed_endpoints": ["/v1/charges"],
            "expires_at": f"{billing_period}-31T23:59:59Z",
        },
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

# Before the billing run:
# 1. Query billing_queue to estimate total charges for this billing period
# 2. Issue a vault key capped at estimated_total * 1.10
# 3. Set stripe.api_key = vault_key (or pass as env var to billing container)
# 4. Run meltano run tap-postgres target-jsonl (or post-target billing lambda)
# 5. The vault key is exhausted after the billing period closes; issue new for next period

The vault key cap addresses the concurrent run scenario specifically: if Run A exhausts the cap while processing 400 customers, Run B’s calls are rejected after the remaining cap is consumed. The audit log in the proxy records which run_id exhausted the cap, which charges succeeded, and at what timestamp — giving you the investigation trail to understand which run billed which customers before the cap was hit.

Governance gaps specific to Meltano

Four Meltano-specific behaviors reduce the effectiveness of billing governance patterns designed for other pipeline tools:

• Singer state is opaque to the billing function. The billing function receives a file of extracted records with no metadata about whether the extraction was a normal incremental run, a failed-then-retried run, or a --no-state-update full-history replay. All three look identical from the billing function’s perspective: a set of JSONL records with the same schema. The pre-flight check must use the billing database as the source of truth for what has already been charged, not any Singer/Meltano metadata field.
• Meltano does not lock the Singer state backend during a run. Two concurrent runs can read the same state and both start from the same bookmark. Meltano provides no built-in at-most-one-pipeline-run gate. Use an external advisory lock (PostgreSQL’s pg_try_advisory_lock, a Redis SET NX EX key, or a billing database row lock) to prevent concurrent billing runs for the same billing period from starting simultaneously.
• meltano schedule run does not check if a previous scheduled run is still active. If the scheduled run takes longer than the schedule interval (for example, a daily billing run that occasionally takes more than 24 hours due to upstream slowness or Stripe rate limiting), Meltano will start the next scheduled run while the previous one is still running. This is the most common source of concurrent runs in Meltano billing pipelines.
• State exclusion patterns applied in development propagate to production via meltano.yml. Unlike a CLI flag (--no-state-update), a state_update_exclusion_patterns setting in meltano.yml persists across deployments and applies to all environments unless explicitly overridden. A pattern added for CI testing that is not scoped to the ci environment in meltano.yml silently disables state for all environments, including production.

Comparison: protection by governance layer

Advisory lock for concurrent run prevention

The pre-flight check closes the duplicate-charge window for customers already in billing_records, but a concurrent run race condition exists during the first run’s commit window — between when Run A has charged a customer and when it commits the billing_records row. Run B’s pre-flight check during that window sees no row for the customer and proceeds to charge. The idempotency key closes this race at the Stripe level (Stripe returns the same charge ID for both calls), but the billing function should also acquire an advisory lock before starting the billing loop to prevent unnecessary Stripe calls:

import psycopg2, os

def run_billing_with_lock(conn, billing_period: str, records: list) -> list:
    cur = conn.cursor()

    # Use PostgreSQL advisory lock scoped to billing_period.
    # pg_try_advisory_lock returns false if another session holds the lock --
    # meaning another billing run for this period is already active.
    lock_key = hash(f"billing-{billing_period}") & 0x7FFFFFFF  # positive int32
    cur.execute("SELECT pg_try_advisory_lock(%s)", (lock_key,))
    acquired = cur.fetchone()[0]

    if not acquired:
        raise RuntimeError(
            f"Another billing run for {billing_period} is already active. "
            "Exiting to prevent concurrent duplicate charges."
        )

    try:
        results = []
        for record in records:
            result = process_usage_record(conn, record)
            results.append(result)
        return results
    finally:
        cur.execute("SELECT pg_advisory_unlock(%s)", (lock_key,))
        conn.commit()

# The meltano run post-hook or billing Lambda entrypoint:
def handler(event, context):
    records = load_records_from_s3(event["s3_key"])
    billing_period = event["billing_period"]
    expected_total = event["expected_total_usd"]

    vault_key = issue_billing_vault_key(billing_period, expected_total)
    stripe.api_key = vault_key  # vault key, not raw Stripe key

    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    try:
        results = run_billing_with_lock(conn, billing_period, records)
    except RuntimeError as e:
        # Concurrent run detected. Log and exit cleanly -- don't retry.
        print(f"BILLING_LOCK_CONTENTION: {e}")
        return {"statusCode": 409, "body": str(e)}

    return {"statusCode": 200, "charged": len([r for r in results if r["status"] == "charged"])}

pytest enforcement suite

import hashlib, pytest
from unittest.mock import patch, MagicMock

# 1. Content-hash idempotency key excludes Singer metadata fields
def test_idempotency_key_excludes_singer_metadata():
    record = {
        "customer_id": "cus_001",
        "amount_cents": 4900,
        "billing_period": "2026-07",
        "_sdc_received_at": "2026-07-17T01:00:00.000000Z",  # must NOT be in key
        "_sdc_sequence": 1721174400000,                       # must NOT be in key
        "_sdc_batched_at": "2026-07-17T01:00:05.000000Z",    # must NOT be in key
    }
    key1 = make_idempotency_key(
        record["customer_id"], record["amount_cents"], record["billing_period"]
    )

    # Simulate a Singer retry: same business event, new Singer metadata
    record_retried = dict(
        record,
        _sdc_received_at="2026-07-17T01:05:00.000000Z",
        _sdc_sequence=1721174700000,
        _sdc_batched_at="2026-07-17T01:05:05.000000Z",
    )
    key2 = make_idempotency_key(
        record_retried["customer_id"],
        record_retried["amount_cents"],
        record_retried["billing_period"],
    )

    assert key1 == key2, "Idempotency key must be stable across Singer retry metadata changes"

# 2. Pre-flight check skips Stripe call for already-billed customer/period
def test_preflight_skips_already_billed(mock_db, mock_stripe):
    mock_db.fetchone.return_value = (1,)  # billing_records row exists

    result = process_usage_record(
        conn=mock_db,
        record={
            "customer_id": "cus_001",
            "stripe_customer_id": "cus_stripe_001",
            "amount_cents": 4900,
            "billing_period": "2026-07",
            "_sdc_received_at": "2026-07-17T01:00:00Z",
        },
    )

    assert result["status"] == "skipped"
    mock_stripe.Charge.create.assert_not_called()

# 3. --no-state-update replay: all historical customers skipped by pre-flight
def test_full_history_replay_all_rows_skipped(mock_db, mock_stripe):
    historical_records = [
        {"customer_id": f"cus_{i}", "stripe_customer_id": f"cus_stripe_{i}",
         "amount_cents": 4900, "billing_period": "2026-01",
         "_sdc_received_at": "2026-07-17T00:00:00Z"}
        for i in range(60)
    ]
    # All 60 customers already have billing_records rows for 2026-01
    mock_db.fetchone.return_value = (1,)

    results = [process_usage_record(mock_db, rec) for rec in historical_records]

    assert all(r["status"] == "skipped" for r in results)
    mock_stripe.Charge.create.assert_not_called()

# 4. Vault key spend cap blocks concurrent-run second wave
def test_vault_key_cap_blocks_concurrent_run(mock_proxy):
    # Expected: 40 customers × $49.00 = $1,960. Vault key capped at $2,156 (110%).
    # Run A charged 40 customers = $1,960. Cap exhausted.
    # Run B (same billing period, concurrent) hits cap immediately.
    mock_proxy.post.side_effect = (
        [MagicMock(status_code=200, json=lambda: {"id": f"ch_{i}"})] * 40
        + [MagicMock(status_code=402, json=lambda: {"error": "spend_cap_exceeded"})] * 40
    )

    # Run A: 40 charges succeed
    run_a = [
        make_charge_via_proxy(f"cus_{i}", 4900, "2026-07", mock_proxy)
        for i in range(40)
    ]
    assert all(r["status_code"] == 200 for r in run_a)

    # Run B (concurrent, same billing period): all blocked after cap exhausted
    run_b = [
        make_charge_via_proxy(f"cus_{i}", 4900, "2026-07", mock_proxy)
        for i in range(40)
    ]
    assert all(r["status_code"] == 402 for r in run_b)

# 5. Advisory lock prevents second concurrent billing run from starting
def test_advisory_lock_blocks_concurrent_run(mock_conn):
    # First call: lock acquired (True)
    # Second call: lock not acquired (False)
    mock_conn.cursor.return_value.fetchone.side_effect = [(True,), (False,)]

    with pytest.raises(RuntimeError, match="already active"):
        run_billing_with_lock(mock_conn, "2026-07", [])

FAQ

Q: Should I use the Singer state backend state or the billing database as the dedup source of truth?
A: Use the billing database. The Singer state backend records where the tap stopped reading the source — it is the tap’s read cursor, not a record of what has been charged. The billing database’s billing_records table is the authoritative record of which customers have been charged for which billing periods. The pre-flight check reads from billing_records, not from the Singer state. This distinction matters because Singer state rollback (failed run) and --no-state-update bypasses the state backend entirely — both failure modes are invisible to a guard that checks Singer state instead of the billing database.

Q: How do I detect when a Meltano run is a --no-state-update replay vs. a normal incremental run?
A: You cannot reliably detect this from the billing function. The Singer target output looks identical for both run types — the same JSONL records with the same schema. The only signal available to a billing function downstream of the target is the volume of records: a --no-state-update run typically emits far more records than a normal incremental run. You can add a record-count gate (if len(records) > expected_incremental_count * 10: raise ValueError(...)) as an early abort, but the pre-flight check is the authoritative protection and handles all cases regardless of record volume.

Q: Can I use Meltano’s built-in deduplication instead of a content-hash idempotency key?
A: Meltano’s stream_map deduplication removes duplicate records within a single extraction run based on a configurable key. It does not deduplicate across runs. A Singer retry re-emits records as new rows in the target (with new _sdc_received_at values) — these rows pass Meltano’s within-run dedup because they are in a different run. The pre-flight check against billing_records is necessary for cross-run dedup.

Q: How long should the vault key live for a Meltano billing pipeline?
A: Issue one vault key per billing period (one per month for monthly billing). Set expires_at to the last second of the billing period (2026-07-31T23:59:59Z for July billing). After the period closes, the key expires and cannot be used for additional charges. Issue a new key for the next billing period. This ensures that a --no-state-update replay of historical data always hits an expired vault key for those periods — the proxy rejects the charge regardless of the pre-flight check outcome.

Q: What do I do if the billing run hits the vault key cap before all customers are charged?
A: The proxy returns a structured 402 spend_cap_exceeded error with the vault key ID and the total spent so far. Query the billing database to identify which customers were successfully charged and which were not. Issue a new vault key for the remaining uncharged amount. Resume the billing run with the new key, relying on the pre-flight check to skip customers who were already charged in the first pass. Do not restart the billing run from the beginning of the customer list — the pre-flight check handles the skipping correctly.

Q: How do I test the advisory lock behavior in a local Meltano development environment?
A: Issue a vault key with "daily_usd_cap": 0. Every Stripe call through that vault key is rejected by the proxy with a 402 spend_cap_exceeded response immediately. Your billing function runs the full path — advisory lock acquisition, pre-flight check, idempotency key generation, vault key call — and each Stripe call is blocked at the proxy. You can verify advisory lock behavior by running two billing function instances simultaneously against the same billing period and confirming that the second instance raises a RuntimeError before making any Stripe calls.