A single webhook registration can feed many internal consumers without sacrificing billing attribution, but only if receipt and consumption are recorded as separate facts. Acknowledge the sender after one durable intake transaction, then require each consumer to acknowledge its own immutable delivery record. For a fintech account platform that issues and revokes one scoped key per tenant, carry tenant_id, event_id, key_id, and consumer_id through every hop. Never infer a tenant later from a queue name, credential, or mutable account lookup.
TL;DR: one public receiver verifies and stores the event, creates one delivery per subscribed consumer, and returns success. Workers claim deliveries with leases and acknowledge them independently. Billing reads the delivery ledger, not application logs or queue depth.
1. How can one webhook registration serve many internal consumers?
The sender needs one answer: did the platform durably accept this event? Internal consumers need a different answer: did this workload finish its assigned copy? Combining those questions creates an attribution trap. A fast audit consumer can make the event look complete while billing has not run, or one slow consumer can force the sender to redeliver work that three siblings completed.
Keep three records. events represents accepted input. deliveries represents the fan-out contract between one event and one consumer. usage_ledger represents billable results, with a uniqueness constraint that makes repeated acknowledgement harmless. The external acknowledgement follows the commit that creates the event and applicable deliveries; it does not wait for model calls or downstream APIs.
This matters during key revocation. An event about key_42 remains attributed to the tenant captured at intake even if that key is disabled before a worker runs. Historical truth must not depend on today's lookup table.
2. Build the smallest runnable ledger first
This Python example uses SQLite so the transitions stay visible. It models an already-authenticated body; production intake must verify the sender before calling receive. Each consumer gets a row, leases it, and acknowledges it independently.
import json
import sqlite3
import time
from contextlib import contextmanager
CONSUMERS = ("risk-evaluator", "billing-meter", "audit-writer")
db = sqlite3.connect("fanout.db")
db.row_factory = sqlite3.Row
db.executescript("""
CREATE TABLE IF NOT EXISTS events (
event_id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL,
key_id TEXT NOT NULL, event_type TEXT NOT NULL,
payload_json TEXT NOT NULL, received_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS deliveries (
event_id TEXT NOT NULL, consumer_id TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'ready', attempt_count INTEGER NOT NULL DEFAULT 0,
lease_until INTEGER, acknowledged_at INTEGER,
PRIMARY KEY (event_id, consumer_id)
);
CREATE TABLE IF NOT EXISTS usage_ledger (
event_id TEXT NOT NULL, consumer_id TEXT NOT NULL, tenant_id TEXT NOT NULL,
units INTEGER NOT NULL CHECK (units >= 0), recorded_at INTEGER NOT NULL,
PRIMARY KEY (event_id, consumer_id)
);
""")
@contextmanager
def transaction():
try:
db.execute("BEGIN IMMEDIATE")
yield
db.commit()
except Exception:
db.rollback()
raise
def receive(body: dict) -> None:
required = {"event_id", "tenant_id", "key_id", "event_type"}
if not required.issubset(body):
raise ValueError("missing attribution field")
now = int(time.time())
with transaction():
inserted = db.execute(
"""INSERT OR IGNORE INTO events VALUES (?, ?, ?, ?, ?, ?)""",
(body["event_id"], body["tenant_id"], body["key_id"],
body["event_type"], json.dumps(body, sort_keys=True), now),
).rowcount
if inserted:
db.executemany(
"INSERT INTO deliveries (event_id, consumer_id) VALUES (?, ?)",
[(body["event_id"], consumer) for consumer in CONSUMERS],
)
def claim(consumer_id: str, lease_seconds: int = 30):
now = int(time.time())
with transaction():
row = db.execute(
"""SELECT d.event_id, e.tenant_id, e.key_id, e.event_type
FROM deliveries d JOIN events e USING (event_id)
WHERE d.consumer_id = ? AND
(d.status = 'ready' OR (d.status = 'processing' AND d.lease_until < ?))
ORDER BY e.received_at LIMIT 1""", (consumer_id, now)
).fetchone()
if row is None:
return None
db.execute(
"""UPDATE deliveries SET status = 'processing', lease_until = ?,
attempt_count = attempt_count + 1
WHERE event_id = ? AND consumer_id = ?""",
(now + lease_seconds, row["event_id"], consumer_id),
)
return dict(row)
def acknowledge(event_id: str, consumer_id: str, units: int) -> None:
now = int(time.time())
with transaction():
event = db.execute(
"SELECT tenant_id FROM events WHERE event_id = ?", (event_id,)
).fetchone()
db.execute(
"INSERT OR IGNORE INTO usage_ledger VALUES (?, ?, ?, ?, ?)",
(event_id, consumer_id, event["tenant_id"], units, now),
)
db.execute(
"""UPDATE deliveries SET status = 'done', acknowledged_at = ?, lease_until = NULL
WHERE event_id = ? AND consumer_id = ?""",
(now, event_id, consumer_id),
)
receive({"event_id": "evt_1007", "tenant_id": "tenant_204",
"key_id": "key_42", "event_type": "scoped_key.revoked"})
job = claim("billing-meter")
if job:
acknowledge(job["event_id"], "billing-meter", units=1)
Run it twice. The second receipt creates neither another event nor another delivery set, and a repeated ledger write does not add units. That is the property an eval should assert.
3. Preserve attribution before expensive work
Billing accuracy starts at intake. Validate the identifiers, verify authenticity, then store the exact attribution tuple beside the payload. A consumer may enrich its result, but it must not overwrite those original identifiers.
I use an eval matrix before adding an agent or retrieval call because those calls increase latency and cost. It has at least six cases: first delivery, duplicate sender delivery, crash before acknowledgement, crash after an external side effect, expired lease, and key revocation while work is queued. For each case, assert delivery counts and ledger totals per tenant. One row of expected integers catches more than a polished dashboard.
There is a hard trade-off. A consumer can make its database update and queue acknowledgement atomic only when both share a transaction. Otherwise, give the side effect an idempotency key such as event_id:consumer_id, then accept redelivery. A durable deduplication key makes the guarantee testable.
The fourth control is making retries boring and poison events visible.
A lease turns a crashed worker from permanent loss into delayed retry. Choose its duration from observed processing latency, then test expiration. Increment attempt_count on every claim so a repeatedly failing payload is distinguishable from ordinary delay.
Do not retry forever. After a bounded policy, quarantine that delivery while retaining its event, tenant, consumer, error category, and timestamps. Operators must be able to replay one consumer without replaying successful siblings.
Fast failures help.
Secrets need a separate lifecycle from payloads. Store verification secrets in a secrets-management system, restrict access, rotate them, and avoid logging them. The OWASP Secrets Management Cheat Sheet recommends documenting and testing rotation and revocation. A scoped key identifier belongs in the attribution record; its secret value does not.
5. Measure the contract instead of queue activity
Queue depth says how much work is waiting. It cannot prove which tenant incurred completed work. Build views from the three-state model: accepted events by tenant, outstanding deliveries by consumer, expired leases, quarantined deliveries, and ledger units by tenant and consumer. Reconcile closed windows so every accepted event has its expected delivery set and every completed billable delivery has one ledger row.
Keep prompts and model output out of the routing envelope unless a consumer needs them. Large envelopes raise storage and transfer cost, while mutable prompt text makes replay comparisons harder. Store a stable payload reference and versioned processing policy for AI workloads. Then an eval can distinguish a routing regression from a prompt regression.
Alert on violated invariants. Two ledger rows for one consumer, a completed delivery without a ledger row, or a tenant mismatch should page sooner than generic worker CPU.
The sixth control is a reconciliation drill, followed by a seventh control: targeted replay.
Before release, create two tenants and two scoped keys, send interleaved events, and stop one worker after its side effect but before acknowledgement. Restart it after the lease expires. Confirm that consumers advance independently, deduplication suppresses repeated effects, and each total stays attached to the tenant captured at intake. Revoke one key while its event waits, then verify that processing retains historical attribution without treating the revoked credential as authorization for a new action.
Finally, rehearse secret rotation and targeted quarantine replay. Record enough metadata to explain a charge without putting credentials or sensitive payload fields in logs. The design is ready when every unit traces to one tenant, one event, and one consumer after retries.
Top comments (0)