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

Pachyderm is a data versioning and pipeline platform built around the concept of versioned data repositories and containerized pipeline steps. Every pipeline job processes “datums” — atomic units of work derived from combinations of input repository commits — in isolated containers. Teams wire Pachyderm pipelines into Stripe billing to close the loop between versioned usage data and SaaS charges: a billing pipeline reads a usage-data repository, processes each customer datum by calling stripe.charges.create(), and writes charge results back to an output repository. Three Pachyderm-specific behaviors introduce Stripe double-charge failure modes that Pachyderm’s own job history, datum logs, and output versioning do not surface as billing risk.

This post covers those three failure modes with Pachyderm pipeline specification JSON, Python transform container code, and the two-layer governance pattern — content-hash idempotency keys plus per-job vault keys via a spend-cap proxy — that eliminates all three without restructuring your pipelines.

Failure mode 1: datum_tries re-runs the billing container from line 1 after any downstream write failure

Pachyderm’s datum_tries pipeline spec field controls how many times a failed datum is retried. The default is 3. When a datum’s container exits with a non-zero status code — whether because Python raised an unhandled exception, a subprocess returned non-zero, or the container was killed — Pachyderm schedules a retry by re-launching the same container image with the same input datum files. There is no partial continuation. The container re-starts at its entrypoint and executes the full transform script from the beginning.

In a billing transform, this means: if stripe.charges.create() succeeds early in the script but the downstream database write raises a psycopg2.OperationalError later, the container exits non-zero. Pachyderm marks the datum as failed and schedules a retry. The retry container re-starts the script from line 1 and calls stripe.charges.create() again — with the same customer, same amount, and without an idempotency key, a new Stripe-generated request ID. A second charge is created. With datum_tries: 3 and a persistent database failure, the customer receives three charges before Pachyderm marks the datum as permanently failed.

# UNSAFE: Pachyderm billing transform with datum_tries but no idempotency key
# A transient database failure after the Stripe call causes Pachyderm to
# retry the datum, re-executing stripe.charges.create() from line 1

import json
import os
import stripe

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

def process_datum(input_path: str, output_path: str):
    with open(f"{input_path}/customer.json") as f:
        customer = json.load(f)

    # Stripe call succeeds -- ch_A created
    charge = stripe.Charge.create(
        amount=customer["amount_cents"],
        currency="usd",
        customer=customer["stripe_customer_id"],
        description=f"Billing for {customer['billing_period']}",
        # no idempotency_key -- Stripe generates a new request ID each call
    )

    # Database write raises psycopg2.OperationalError ->
    # container exits non-zero -> Pachyderm retries the datum ->
    # next container run creates ch_B for the same customer + billing_period
    write_charge_to_db(
        customer_id=customer["id"],
        charge_id=charge["id"],
        billing_period=customer["billing_period"],
    )

    with open(f"{output_path}/result.json", "w") as f:
        json.dump({"customer_id": customer["id"], "charge_id": charge["id"]}, f)

if __name__ == "__main__":
    process_datum("/pfs/usage_data", "/pfs/out")

Pachyderm’s datum log shows the datum moving through starting → running → failed → starting → running → failed → starting → running → failed. There is no billing-specific signal in the datum state machine — Pachyderm does not know that stripe.charges.create() succeeded before the exception. Stripe’s dashboard shows three ch_ objects for the same customer in the same billing period, each with a different request ID and a different charge creation timestamp. The first two are orphaned: they have no corresponding row in the billing records table.

# SAFE: Pachyderm billing transform with content-hash idempotency key
# The same idempotency key on every datum retry causes Stripe to return
# the existing charge rather than creating a new one

import hashlib
import json
import os
import stripe

VAULT_KEY = os.environ["KEYBRAKE_VAULT_KEY"]   # per-job vault key from Keybrake
PROXY_BASE = "https://proxy.keybrake.com/stripe"

def process_datum(input_path: str, output_path: str):
    with open(f"{input_path}/customer.json") as f:
        customer = json.load(f)

    # Content-hash idempotency key: stable across all datum retries
    # Does NOT include Pachyderm job ID or datum ID -- those change on retry
    raw_key = (
        f"{customer['id']}:{customer['amount_cents']}"
        f":{customer['billing_period']}:pachyderm-billing"
    )
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    stripe.api_key  = VAULT_KEY
    stripe.api_base = PROXY_BASE

    try:
        charge = stripe.Charge.create(
            amount=customer["amount_cents"],
            currency="usd",
            customer=customer["stripe_customer_id"],
            description=f"Billing for {customer['billing_period']}",
            idempotency_key=idempotency_key,  # SAFE: same key on every datum retry
        )
    except stripe.error.CardError:
        result = {"customer_id": customer["id"], "charge_id": None, "error": "card_declined"}
        with open(f"{output_path}/result.json", "w") as f:
            json.dump(result, f)
        return
    except stripe.error.InvalidRequestError as e:
        result = {"customer_id": customer["id"], "charge_id": None, "error": str(e)}
        with open(f"{output_path}/result.json", "w") as f:
            json.dump(result, f)
        return

    # If the database write fails and Pachyderm retries the datum,
    # the next container run calls stripe.Charge.create() with the same
    # idempotency_key -> Stripe returns the original ch_A, no new charge
    write_charge_to_db(
        customer_id=customer["id"],
        charge_id=charge["id"],
        billing_period=customer["billing_period"],
    )

    with open(f"{output_path}/result.json", "w") as f:
        json.dump({"customer_id": customer["id"], "charge_id": charge["id"]}, f)

if __name__ == "__main__":
    process_datum("/pfs/usage_data", "/pfs/out")

The critical constraint: the idempotency key must be derived from the business inputs — customer ID, amount, billing period — not from Pachyderm’s job ID, datum ID, or input commit hash. Those identifiers change on each retry attempt. A key derived from the datum ID is a new key to Stripe on each retry and provides no deduplication.

Failure mode 2: parallelism_spec.constant: N spawns N concurrent worker pods sharing one unrestricted STRIPE_SECRET_KEY

Pachyderm’s parallelism_spec.constant field controls how many worker pods are created for a pipeline job. When set to N, Pachyderm launches N Kubernetes pods simultaneously, each processing a different subset of the job’s datums. All N pods receive the same environment from the pipeline spec — including the same STRIPE_SECRET_KEY mounted from a Kubernetes Secret. There is no per-pod spend cap. There is no rate limit per worker. All N workers independently call stripe.charges.create() for their assigned datums concurrently.

The failure scenario: a data pipeline upstream of your billing repository has a bug. The amount_cents field in the usage data files is stored in dollars, not cents — so a customer whose plan costs $199.00/month has amount_cents: 199 instead of 19900. Or the field is correctly in cents in the raw data, but a dbt transformation upstream added an accidental × 100 conversion. Your billing pipeline has parallelism_spec.constant: 8 and 600 customer datums. Pachyderm launches 8 worker pods, each processing 75 datums. All 8 pods call stripe.charges.create() for their 75 customers before any result is written to the output repository. All 600 charges succeed — Stripe accepts any positive integer as a valid charge amount. By the time you inspect the output repository and notice that all amount_cents values are exactly 100× too small, all 600 incorrect charges have been submitted and settled.

# pipeline spec: UNSAFE parallelism configuration
# All N worker pods share one STRIPE_SECRET_KEY with no per-pod spend cap
{
  "pipeline": { "name": "billing_pipeline" },
  "description": "Monthly billing pipeline",
  "transform": {
    "image": "myorg/billing-transform:latest",
    "cmd": ["python", "/app/transform.py"],
    "env": {
      "STRIPE_SECRET_KEY": "sk_live_..."
    }
  },
  "parallelism_spec": { "constant": 8 },
  "input": {
    "pfs": {
      "repo": "usage_data",
      "branch": "master",
      "glob": "/customers/*"
    }
  },
  "datum_tries": 3
}

# pipeline spec: SAFE parallelism configuration
# Per-job vault key from Keybrake caps total spend across all worker pods
# A data error causes the cap to be hit and the proxy rejects excess charges
{
  "pipeline": { "name": "billing_pipeline" },
  "description": "Monthly billing pipeline",
  "transform": {
    "image": "myorg/billing-transform:latest",
    "cmd": ["python", "/app/transform.py"],
    "secrets": [
      {
        "name": "keybrake-credentials",
        "key": "KEYBRAKE_API_KEY",
        "env_var": "KEYBRAKE_API_KEY"
      }
    ]
  },
  "parallelism_spec": { "constant": 8 },
  "input": {
    "pfs": {
      "repo": "usage_data",
      "branch": "master",
      "glob": "/customers/*"
    }
  },
  "datum_tries": 3
}

The vault key is issued once per billing job before the workers start — not inside each datum’s transform container. The cleanest pattern is an upstream pipeline step that reads the billing manifest, issues a single vault key capped at 110% of the expected total, and writes it to an output repository that the billing pipeline reads as a second input. Each billing datum worker reads the vault key from its input files and uses it as stripe.api_key against the proxy.

# vault_key_pipeline transform: issues one vault key per billing job
# upstream of the billing pipeline

import json
import os
import httpx

KEYBRAKE_API_KEY = os.environ["KEYBRAKE_API_KEY"]

def issue_vault_key(input_path: str, output_path: str):
    # Read the billing manifest to compute expected total
    with open(f"{input_path}/manifest.json") as f:
        manifest = json.load(f)

    billing_period  = manifest["billing_period"]
    expected_total  = manifest["expected_total_cents"]
    customer_count  = manifest["customer_count"]

    resp = httpx.post(
        "https://api.keybrake.com/vault-keys",
        headers={"Authorization": f"Bearer {KEYBRAKE_API_KEY}"},
        json={
            "label": f"pachyderm-billing-{billing_period}",
            "vendor": "stripe",
            "spend_cap_usd": round(expected_total / 100 * 1.10, 2),
            "ttl_seconds": 3600,
            "allowed_endpoints": ["/v1/charges"],
        },
        timeout=10,
    )
    resp.raise_for_status()
    vault_key = resp.json()["vault_key"]

    with open(f"{output_path}/vault.json", "w") as f:
        json.dump({"vault_key": vault_key, "billing_period": billing_period}, f)

if __name__ == "__main__":
    issue_vault_key("/pfs/billing_manifest", "/pfs/out")

Each billing datum worker reads /pfs/vault_keys/vault.json, extracts the vault_key, and uses it as stripe.api_key. All 8 pods share the same vault key, but the proxy enforces the total spend cap across all concurrent calls from all pods. A data error that would cause total charges to exceed 110% of the expected amount causes the proxy to return 402 Payment Required after the cap is reached, not after all 600 charges go through.

Failure mode 3: pachctl run pipeline and a cron-triggered job process the same input commit simultaneously

Pachyderm’s cron input type creates a new commit to a special cron repository on each scheduled tick. Each new commit triggers a pipeline job. If the pipeline is still processing the previous cron tick’s commit when the next tick fires, Pachyderm queues a new job for the new commit. These jobs are sequential by default — but there are two common scenarios that cause two concurrent jobs to process the same billing-period datums simultaneously.

The first scenario: an operator runs pachctl run pipeline billing_pipeline after a partial job failure. This command creates a new job for the most recent input commit — the same commit the failed job was processing. If Pachyderm has already auto-retried the failed datums and the job is in a running or merging state, both the original job and the operator-triggered job can be active simultaneously. Both spawn worker pods over the same datum set. Both call stripe.charges.create() for the same customers.

The second scenario: an operator manually commits a billing trigger file to the usage data repository — for example, by running pachctl put file usage_data@master -f billing_trigger.json — while a cron-triggered job for the same billing period is already running. Pachyderm creates a new commit to the usage_data repository, which triggers a new pipeline job. Both jobs see the same customer datums in the input. Both schedule stripe.charges.create() for all customers.

# What Pachyderm's job list looks like during a double-execution incident:
#
# $ pachctl list job --pipeline billing_pipeline
#
# ID                               PIPELINE         STARTED        DURATION  RESTART  PROGRESS   DL       UL       STATE
# a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4 billing_pipeline 8 minutes ago  -         0        450/600    1.2MiB   890KiB   running
# f6e5d4c3b2a1f6e5d4c3b2a1f6e5d4c3 billing_pipeline 2 minutes ago  -         0        120/600    320KiB   240KiB   running
#
# Both jobs are processing the same input commit (usage_data@master).
# Job a1b2c3... has processed 450 customer datums and called Stripe for each.
# Job f6e5d4... has processed 120 customer datums, 80 of which overlap with a1b2c3...
# Pachyderm's datum cache prevents f6e5d4... from re-processing a datum only
# if a1b2c3... completed it -- running datums are not cached.
# Customers in the overlap zone are being charged twice simultaneously.

Pachyderm does have a datum result cache: when a job for a given commit completes a datum, the result is cached, and a subsequent job for the same commit with the same datum inputs skips re-processing that datum. But the cache only activates once the datum is completed. If a datum is in progress when the second job starts, the cache miss causes the second job to also schedule that datum. Two worker containers from two different jobs both execute the billing transform for the same customer.

# SAFE: billing transform with a per-billing-period advisory lock
# and content-hash idempotency key that closes both failure scenarios

import hashlib
import json
import os
import psycopg2
import stripe

VAULT_KEY  = None   # populated from /pfs/vault_keys/vault.json at startup
PROXY_BASE = "https://proxy.keybrake.com/stripe"
DB_URL     = os.environ["DATABASE_URL"]

def load_vault_key() -> str:
    with open("/pfs/vault_keys/vault.json") as f:
        return json.load(f)["vault_key"]

def try_acquire_datum_lock(conn, customer_id: str, billing_period: str) -> bool:
    """Advisory lock: only one worker processes this (customer, billing_period) pair."""
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO billing_datum_locks (customer_id, billing_period, locked_at)
            VALUES (%s, %s, NOW())
            ON CONFLICT (customer_id, billing_period) DO NOTHING
            """,
            (customer_id, billing_period),
        )
        conn.commit()
        return cur.rowcount == 1  # True = this worker acquired the lock

def process_datum(input_path: str, output_path: str):
    vault_key = load_vault_key()

    with open(f"{input_path}/customer.json") as f:
        customer = json.load(f)

    conn = psycopg2.connect(DB_URL)

    # Advisory lock: if another job already locked this datum, skip it
    if not try_acquire_datum_lock(conn, customer["id"], customer["billing_period"]):
        result = {
            "customer_id": customer["id"],
            "charge_id": None,
            "skipped": True,
            "reason": "datum_lock_already_held",
        }
        with open(f"{output_path}/result.json", "w") as f:
            json.dump(result, f)
        return

    # Content-hash idempotency key: stable across any re-execution
    raw_key = (
        f"{customer['id']}:{customer['amount_cents']}"
        f":{customer['billing_period']}:pachyderm-billing"
    )
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    stripe.api_key  = vault_key
    stripe.api_base = PROXY_BASE

    try:
        charge = stripe.Charge.create(
            amount=customer["amount_cents"],
            currency="usd",
            customer=customer["stripe_customer_id"],
            description=f"Billing for {customer['billing_period']}",
            idempotency_key=idempotency_key,
        )
    except stripe.error.CardError:
        result = {"customer_id": customer["id"], "charge_id": None, "error": "card_declined"}
        with open(f"{output_path}/result.json", "w") as f:
            json.dump(result, f)
        return
    except stripe.error.InvalidRequestError as e:
        result = {"customer_id": customer["id"], "charge_id": None, "error": str(e)}
        with open(f"{output_path}/result.json", "w") as f:
            json.dump(result, f)
        return

    write_charge_to_db(
        conn=conn,
        customer_id=customer["id"],
        charge_id=charge["id"],
        billing_period=customer["billing_period"],
    )

    with open(f"{output_path}/result.json", "w") as f:
        json.dump({"customer_id": customer["id"], "charge_id": charge["id"]}, f)

if __name__ == "__main__":
    process_datum("/pfs/usage_data", "/pfs/out")

The advisory lock is a two-column unique constraint on (customer_id, billing_period) in a billing_datum_locks table. The first worker to insert a row for a given (customer_id, billing_period) pair proceeds with the Stripe charge. Any subsequent worker — from a concurrent job, a duplicate trigger, or a datum retry that somehow bypassed the idempotency key — finds the row already present and writes a skipped result to the output repository. Pachyderm records the skipped datum as successful (zero exit), and the billing audit log shows which customers were protected by the lock rather than silently swallowed by a cache miss.

Gap analysis: four more Pachyderm–Stripe failure modes to address

The three failure modes above cover the most common Pachyderm billing incidents. Four additional scenarios are worth closing before going to production:

1. Input glob over multiple files per datum double-counts charges. A glob pattern of /customers/* creates one datum per file in the customers/ directory. But a glob of /customers/ (no wildcard) creates one datum for the entire directory — all customer files together. If the upstream repository is restructured and the glob changes from per-file to per-directory, your transform suddenly processes all customers in a single datum, but the idempotency keys derived from individual customer IDs still work correctly. The danger is the reverse: if you previously had a per-directory datum and split to per-file, existing billing period records in the advisory lock table from the per-directory run cover only the directory-level lock, not per-customer locks. Run a billing-period audit after any glob change.

2. --reprocess flag bypasses Pachyderm’s datum cache and re-runs all datums. pachctl run pipeline billing_pipeline --reprocess ignores all cached datum results and re-runs every datum from scratch. This is useful for debugging transform logic changes but is catastrophic for billing pipelines: all customers in the billing period are re-processed, and without idempotency keys, all are charged again. With content-hash idempotency keys, Stripe returns the original charge for each customer. With the advisory lock, the re-run workers find existing lock rows and write skipped results. Both layers are required: the idempotency key protects against database failures; the advisory lock protects against the --reprocess flag bypassing the idempotency key’s 24-hour window.

3. Vault key TTL must cover the full job wall-clock time including retries. The vault key is issued once per billing job before datum workers start. With parallelism_spec.constant: 8 and 600 datums, a healthy job might take 15 minutes. But with datum_tries: 3 and transient database failures, a job might run for 90 minutes before all datums complete. If the vault key’s TTL is set to 30 minutes, workers that start processing datums after the 30-minute mark receive AuthenticationError from the proxy. Set the TTL to at least 3× the expected wall-clock time, and catch AuthenticationError explicitly in the transform to return a structured error dict rather than exiting non-zero (which would trigger another datum retry with an expired vault key).

4. Pachyderm Enterprise egress pipeline leaks vault keys into output commits. If your billing pipeline has an egress configuration that copies output files to S3 or another external store, and your output result.json includes the vault key value for debugging, those keys are written to the egress destination and may persist after the vault key expires. Write only charge IDs and status fields to output files. Keep vault keys in environment variables or input-only repository files that are not written to the output repository.

Approach comparison

Enforcement test suite

Five tests that should pass before a Pachyderm billing pipeline reaches production:

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

# --- Test 1: idempotency key is stable across datum retries ---

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

def test_idempotency_key_stable_across_retries():
    k1 = make_idempotency_key("cust_123", 14995, "2026-07")
    k2 = make_idempotency_key("cust_123", 14995, "2026-07")
    assert k1 == k2  # same inputs -> same key on every datum retry

# --- Test 2: idempotency key isolates billing periods ---

def test_idempotency_key_period_isolation():
    k_july = make_idempotency_key("cust_123", 14995, "2026-07")
    k_aug  = make_idempotency_key("cust_123", 14995, "2026-08")
    assert k_july != k_aug  # same customer, different period -> different key

# --- Test 3: CardError is returned as a result dict, not raised ---

def test_card_error_returns_dict_not_raises():
    import stripe
    with patch("stripe.Charge.create", side_effect=stripe.error.CardError(
        "Your card was declined.", None, "card_declined", http_status=402, http_body=None
    )):
        with patch("builtins.open", MagicMock()):
            stripe.api_key  = "vk_test_..."
            stripe.api_base = "https://proxy.keybrake.com/stripe"
            try:
                charge = stripe.Charge.create(
                    amount=14995, currency="usd", customer="cus_test",
                    idempotency_key="abc123",
                )
            except stripe.error.CardError as e:
                result = {"error": "card_declined"}
            assert result["error"] == "card_declined"

# --- Test 4: advisory lock prevents second worker from billing same customer ---

def test_advisory_lock_blocks_second_worker(tmp_path):
    import psycopg2
    # Simulate: first worker inserts lock row (rowcount=1 -> proceed)
    # Second worker insert returns rowcount=0 -> skip
    mock_cursor_1 = MagicMock()
    mock_cursor_1.rowcount = 1   # first worker acquires lock
    mock_cursor_2 = MagicMock()
    mock_cursor_2.rowcount = 0   # second worker finds row already present

    conn_1 = MagicMock()
    conn_1.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor_1)
    conn_1.cursor.return_value.__exit__  = MagicMock(return_value=False)

    conn_2 = MagicMock()
    conn_2.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor_2)
    conn_2.cursor.return_value.__exit__  = MagicMock(return_value=False)

    def try_acquire(conn, customer_id, billing_period):
        with conn.cursor() as cur:
            return cur.rowcount == 1

    assert try_acquire(conn_1, "cust_123", "2026-07") is True   # first worker proceeds
    assert try_acquire(conn_2, "cust_123", "2026-07") is False  # second worker skips

# --- Test 5: idempotency key excludes all Pachyderm runtime identifiers ---

def test_idempotency_key_excludes_runtime_ids():
    import uuid
    job_id_1   = str(uuid.uuid4())
    job_id_2   = str(uuid.uuid4())
    datum_id_1 = str(uuid.uuid4())
    datum_id_2 = str(uuid.uuid4())

    # Key derived purely from business inputs -- runtime IDs not included
    k1 = make_idempotency_key("cust_123", 14995, "2026-07")
    k2 = make_idempotency_key("cust_123", 14995, "2026-07")

    # Different Pachyderm job IDs and datum IDs must not produce different keys
    assert k1 == k2  # business inputs unchanged -> idempotency key unchanged

Frequently asked questions

Can I use the Pachyderm job ID or datum ID as the Stripe idempotency key?

No. Pachyderm assigns a new job ID to every job, including retried jobs triggered by datum_tries. The datum ID is stable for a given datum within a job, but a new job for the same input commit gets a new set of datum IDs. An idempotency key derived from either identifier is a new key to Stripe on each retry or job re-run. Use the business-input hash — sha256(customer_id:amount_cents:billing_period:pachyderm-billing) — which is stable regardless of how many times Pachyderm reschedules the datum.

Does Pachyderm’s datum cache make idempotency keys redundant?

No. The datum cache prevents a completed datum from being re-processed by a subsequent job for the same input commit. It does not protect against two concurrent jobs both processing an in-progress datum, nor does it protect against a --reprocess flag that explicitly bypasses the cache, nor against Stripe’s 24-hour idempotency window expiry for billing periods that span multiple months. Idempotency keys and the datum cache are complementary: the cache reduces redundant re-processing at the Pachyderm layer, while idempotency keys protect at the Stripe layer regardless of how the Pachyderm job was triggered.

How should I size the vault key TTL for a large billing pipeline?

Measure your pipeline’s p99 wall-clock time across several billing cycles, then set the TTL to 3× that value, with a minimum of 3600 seconds (one hour) for any pipeline with datum_tries above 1. For a pipeline with 600 datums, parallelism_spec.constant: 8, and datum_tries: 3, the worst-case runtime is approximately (600 / 8) × 3 × datum_avg_seconds + retry_delays. Size generously: a vault key that expires mid-job causes AuthenticationError on active workers, triggers datum retries for the expired-key failures, and can create a retry cascade that burns through datum_tries on datums that would otherwise complete cleanly.

What happens if the vault key pipeline step fails before issuing a key?

Make the vault key issuance step a separate upstream pipeline with its own output repository that the billing pipeline reads as a second input. If the vault key pipeline fails, Pachyderm does not trigger the downstream billing pipeline for that commit. This is safer than issuing the vault key inside the billing transform itself, where each worker pod would attempt to issue its own key and could race each other. One vault key per billing job, issued before datum workers start, with the key distributed via a read-only input repository that billing workers cannot write to.

Does the advisory lock table need to be cleaned up between billing periods?

Rows in billing_datum_locks for past billing periods can be archived or deleted after the billing period’s reconciliation window closes (typically 30–90 days after the billing date). The unique constraint is on (customer_id, billing_period), so rows from different billing periods do not conflict. A nightly cleanup job that deletes rows older than 90 days keeps the table small without removing rows that might still be referenced by late-arriving dispute reconciliations.

Is the Keybrake proxy compatible with Pachyderm Enterprise’s egress features?

Yes. The Keybrake proxy is a standard HTTPS reverse proxy that intercepts calls to proxy.keybrake.com/stripe/v1/... and forwards them to api.stripe.com/v1/... after enforcing spend caps and logging. Your Pachyderm transform containers call the proxy endpoint the same way they would call Stripe’s API directly, using the vault key as stripe.api_key and the proxy base URL as stripe.api_base. Pachyderm Enterprise egress, pipeline metadata, and S3 output repositories are unaffected. The proxy’s audit log provides a full record of which Pachyderm jobs and datum workers called Stripe and at what cost, independent of Pachyderm’s own job history.