Building a Highly Concurrent Webhook Processing Pipeline: Lessons from the Synapse Reconciliation Engine

Adrian obungu — Wed, 24 Jun 2026 20:06:38 +0000

Financial webhooks are deceptively simple on the surface. A third-party payment gateway sends an HTTP POST to your endpoint. You process it. You respond. The transaction is recorded.

In practice, this description conceals a category of engineering problems that only reveal themselves under production conditions: duplicate delivery during network instability, malformed payloads from legacy client versions, floating-point arithmetic errors in financial calculations, and cascading failures when a downstream compliance API degrades. Each of these failure modes, left unaddressed, produces outcomes ranging from duplicate tax invoices to silent data corruption in a financial ledger.

The Synapse Reconciliation Engine was built to bridge two of East Africa's most critical financial infrastructure components — Safaricom's M-Pesa Daraja API and the Kenya Revenue Authority's eTIMS compliance gateway. This article documents the architectural decisions made at each stage of the ingress-to-compliance pipeline, with a focus on the concurrency model, schema integrity, and graceful degradation strategy that define its production posture.

*Ingress & Idempotency
*
*The Fast-Acknowledgement Contract
*
Safaricom's Daraja API imposes a hard constraint on webhook consumers: the callback endpoint must return an HTTP 200 OK within a narrow timeout window. If the response is delayed — by a slow database write, a blocked event loop, or an outbound API call — Daraja treats the delivery as failed and retries. Under network instability, this retry behaviour compounds: a single payment event can generate three, five, or more duplicate callback deliveries in rapid succession.

The naive architectural response is to handle deduplication at the database layer. This is incorrect. By the time a duplicate reaches the database, the application has already consumed event loop time parsing the payload, acquired a connection from the Postgres pool, and potentially triggered an outbound call to the KRA eTIMS API. The cost of the duplicate has already been paid.

The correct response is to enforce the idempotency guarantee at the ingress boundary, before any downstream I/O occurs.

Synapse implements this using a Redis SET NX (set if not exists) command with a 24-hour TTL, keyed on Daraja's CheckoutRequestID — a globally unique identifier assigned per STK Push transaction. The operation is atomic by design: Redis guarantees that only one caller can set a given key, regardless of concurrent arrivals. If the key already exists, the callback is a duplicate and is dropped in sub-millisecond time. The HTTP 200 is returned immediately in both cases, satisfying Daraja's acknowledgement contract without triggering any downstream processing.

`Python

@router.post("/callback")
async def mpesa_callback(
payload: MpesaWebhookPayload,
background_tasks: BackgroundTasks,
request: Request
):
checkout_id = payload.Body.stkCallback.CheckoutRequestID
is_novel = await request.app.state.storage.check_idempotency(checkout_id)

if not is_novel:
    return {"ResultCode": 0, "ResultDesc": "Duplicate ignored"}

background_tasks.add_task(process_compliance_pipeline, payload, request)
return {"ResultCode": 0, "ResultDesc": "Accepted"}`

The BackgroundTasks dispatch is the architectural boundary between the acknowledgement contract and the compliance pipeline. The event loop returns the HTTP 200 before the background task begins execution, which means Daraja's timeout window is never at risk regardless of downstream latency.

*Structural Integrity at the Boundary
*
Before idempotency is checked, the payload must be structurally valid. Pydantic v2 enforces this at the function signature level. Any payload that fails schema validation — missing required fields, incorrect types, or malformed nested structures — is rejected with an HTTP 422 Unprocessable Entity before it reaches the application layer. This is not merely a convenience; it is a security boundary. Unvalidated financial payloads reaching business logic are a source of both data corruption and potential injection vectors.

*Schema Transformation
*
*The E.164 Normalisation Problem
*
Daraja's STK Push webhook delivers the customer's phone number as a metadata item within a dynamic key-value array. The format of this value is inconsistent across client versions and network conditions. Observed formats include 0712345678, +254712345678, 254712345678, 0112345678, and integer-coerced floats such as 254712345678.0 — a consequence of JSON serialisers treating numeric strings as numbers in some SDK implementations.

Passing any of these raw values downstream creates two problems. First, the KRA eTIMS schema requires a strict E.164 format (254...). Second, normalising the value inside an async coroutine using an uncompiled regex introduces overhead on every invocation — the regex engine must reparse the pattern string each time, which under high concurrency creates measurable event loop contention.

Synapse addresses both problems with a pre-compiled regex pattern, evaluated once at module import time:

`Python

PHONE_REGEX = re.compile(r"^(?:(?:+?254)|0)?([17]\d{8})$")

def safe_normalize_phone(raw_phone: Any, fallback: str = "254700000000") -> str:
val_str = str(raw_phone).strip().replace(" ", "").replace("-", "")
if val_str.endswith(".0"):
val_str = val_str[:-2]
match = PHONE_REGEX.match(val_str)
return "254" + match.group(1) if match else fallback`

The **.endswith(".0") **strip handles the float-coercion edge case before the regex is applied. The fallback value ensures the pipeline does not abort on an unparseable phone number — a deliberate trade-off that prioritises ledger completeness over strict rejection, since the transaction itself is valid even if the phone number is malformed.

*Financial Precision and the Decimal Boundary
*
Daraja's Amount field is delivered as a JSON number. In Python, JSON numbers without explicit decimal notation are parsed as float. This is a correctness problem for financial systems. IEEE 754 floating-point arithmetic cannot represent all decimal fractions exactly — 0.1 + 0.2 evaluates to 0.30000000000000004. At scale, these rounding errors accumulate into ledger discrepancies that are difficult to audit and impossible to explain to a regulator.

The Synapse schema transformer converts the Amount field to Decimal at the boundary, using Python's decimal module which provides arbitrary-precision fixed-point arithmetic:

`Python

from decimal import Decimal, ROUND_HALF_UP

amount_decimal = Decimal(str(raw_amount)).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP
)`

The conversion via str() is deliberate. Passing a float directly to Decimal() would inherit the floating-point imprecision. Converting to string first forces the decimal module to parse the human-readable representation, which is exact.

The ROUND_HALF_UP rounding mode aligns with standard accounting conventions and KRA's invoice precision requirements.

*Mapping Dynamic Arrays to a Strict Schema
*
Daraja's CallbackMetadata.Item is a list of key-value objects rather than a fixed schema. Fields like Amount, MpesaReceiptNumber, and PhoneNumber must be extracted by iterating the array and matching on the Name key. This dynamic structure is incompatible with the strict, typed payload required by the eTIMS saveVscu endpoint.

The ETIMSTransformer class encapsulates this mapping logic, producing a fully validated Pydantic model as output:

`Python

@staticmethod
def transform_daraja_to_etims(payload: MpesaWebhookPayload) -> ETIMSInvoicePayload:
items = {
item.Name: item.Value
for item in payload.Body.stkCallback.CallbackMetadata.Item
}
return ETIMSInvoicePayload(
invoiceNumber=items["MpesaReceiptNumber"],
totalAmount=Decimal(str(items["Amount"])).quantize(Decimal("0.01")),
phoneNumber=safe_normalize_phone(items.get("PhoneNumber", "")),
transactionDate=datetime.utcnow().isoformat()
)
`

The transformer is a pure function with no side effects, which makes it trivially testable in isolation from the rest of the pipeline.

*Graceful Degradation
*
*The Shared Connection Pool
*
Outbound HTTP calls to the KRA eTIMS API are the highest-latency operation in the pipeline. Under high concurrency, naive instantiation of a new httpx.AsyncClient per request would exhaust the system's file descriptor limit and saturate the TLS handshake overhead. Synapse avoids this by instantiating a single shared httpx.AsyncClient during the FastAPI lifespan context, configured with explicit connection pool limits:

`Python

limits = httpx.Limits(
max_connections=200,
max_keepalive_connections=100,
keepalive_expiry=30.0
)
http_client = httpx.AsyncClient(limits=limits, timeout=httpx.Timeout(10.0 ))`

The keepalive_expiry of 30 seconds ensures that idle connections are not held open indefinitely, while max_keepalive_connections=100 maintains a warm pool for burst traffic — avoiding TLS handshake overhead on repeated calls to the same endpoint.

*Concurrency Capping with asyncio.Semaphore
*
A shared connection pool prevents file descriptor exhaustion, but it does not prevent the application from flooding the KRA API with more concurrent requests than the downstream service can handle. An asyncio.Semaphore provides a soft concurrency cap at the application layer:

`Python

ETIMS_SEMAPHORE = asyncio.Semaphore(50)

async def submit_to_etims(payload: ETIMSInvoicePayload, client: httpx.AsyncClient ):
async with ETIMS_SEMAPHORE:
# outbound call executes here`

The semaphore value of 50 is a deliberate configuration choice: it is low enough to protect the downstream API from being overwhelmed, but high enough to sustain meaningful throughput during peak transaction windows. This value should be tuned against observed KRA API rate limits once production credentials are available.

*Exponential Backoff with Jitter
*
Transient failures from the KRA eTIMS API — network timeouts, 502 responses during deployments, TLS renegotiation errors — are expected in any production environment. A naive retry loop without delay creates a retry storm: all failed requests retry simultaneously, amplifying load on an already-degraded service.

Synapse implements exponential backoff with full jitter on the retry loop:

Python for attempt in range(1, MAX_RETRIES + 1): try: response = await client.post(ETIMS_URL, json=payload.model_dump()) response.raise_for_status() return except (httpx.RequestError, httpx.HTTPStatusError ) as exc: if attempt == MAX_RETRIES: logger.error("eTIMS submission permanently failed", exc_info=exc) raise sleep_duration = (2 ** attempt) + random.uniform(0, 1) await asyncio.sleep(sleep_duration)

The random.uniform(0, 1) jitter term desynchronises retries across concurrent coroutines. Without it, all coroutines that failed at the same moment would retry at the same moment — recreating the spike that caused the failure in the first place. The jitter term spreads retries across a one-second window, smoothing the load curve on the downstream service.

*Enterprise Readiness
*
The decisions documented here — atomic idempotency at ingress, pre-compiled normalisation, Decimal precision at the financial boundary, shared connection pooling, semaphore-capped concurrency, and jittered backoff — are not independent optimisations. They form a coherent philosophy: enforce guarantees as early as possible, minimise the blast radius of any single failure, and design every component to degrade gracefully rather than catastrophically.

A system that drops duplicate webhooks silently, normalises dirty data without aborting, and retries compliance submissions with controlled backoff is not merely more reliable than one that does not. It is qualitatively different — it is a system that can be operated at scale, audited after an incident, and extended without fear of hidden coupling.

The Synapse Reconciliation Engine is not yet at full production deployment. The remaining gaps — merchant authentication, a dead letter queue for permanently failed eTIMS submissions, and live KRA sandbox validation — are well-understood and bounded. The architectural foundation, however, is production-grade. That distinction matters when the system processes financial transactions that carry tax compliance obligations.

_The Synapse Reconciliation Engine is an open architecture middleware layer for the East African digital payments ecosystem. The repository is available at github.com/Adrian-Obungu/synapse-reconciliation-engine.
_

DEV Community: Adrian obungu

Building a Highly Concurrent Webhook Processing Pipeline: Lessons from the Synapse Reconciliation Engine