Handling Rate Limits in Salesforce API Sync

Within the broader Cross-System Data Discovery & Sync architecture, the Salesforce connector is one of the first vendors to throttle under Data Subject Request (DSR) load — and a stalled connector is a compliance exposure, not a network hiccup. This page sits inside the SaaS API Sync Strategies layer and covers the specific problem a privacy data engineer hits when a subject’s Salesforce records must be extracted before a regulatory deadline but the org’s REST and Bulk APIs return 429 Too Many Requests. Salesforce enforces throttling across orthogonal dimensions — a rolling 24-hour daily allocation, short-lived concurrent (burst) limits, and Bulk API batch ceilings — and a naive retry loop makes every one of them worse. The pattern below treats throttling as a predictable state transition: parse the limit telemetry, back off deterministically, keep extracted PII encrypted while you wait, enforce idempotency on writes, and dead-letter anything that cannot complete inside the statutory window.

Because neither the GDPR Article 12(3) one-month deadline nor the CCPA §1798.130(a)(2) 45-day baseline pauses for a vendor’s rate limiter, the technical retry ceiling must be sized as a fraction of the compliance SLA, and exhaustion must escalate to a human rather than drop silently.

Prerequisites #

• Python 3.11+ — for zoneinfo, tomllib, and precise time.monotonic() semantics.
• requests 2.31+ — HTTP session with connection pooling for the REST endpoints; the same pattern applies to simple-salesforce if you prefer a typed client.
• pydantic 2.x — typed parsing of limit telemetry and request envelopes using ConfigDict and field_validator (v2 idioms only).
• cryptography 42+ — Fernet (AES-128-CBC + HMAC) for at-rest encryption of transient PII buffers.
• A Salesforce Connected App with a dedicated integration user scoped to read-only DSR discovery, so the connector never exceeds the least-privilege boundary the SaaS API Sync Strategies cluster mandates.
• A durable queue (Redis, SQS, or Kafka) fronting the dead-letter path; the in-process queue shown here is illustrative and must be backed by persistent storage in production.

Step-by-step implementation #

Step 1 — Parse limit telemetry into a typed model #

Salesforce reports rolling 24-hour consumption on the Sforce-Limit-Info header of every response (format api-usage=1234/5000000). Retry-After is only present for some limit types and is absent during burst throttling, so the parser must treat it as optional. Modeling the telemetry with Pydantic v2 gives you validation and a single source of truth for downstream decisions.

from __future__ import annotations

import re
from pydantic import BaseModel, ConfigDict, field_validator
from requests.structures import CaseInsensitiveDict

_USAGE_RE = re.compile(r"api-usage=(?P<used>\d+)/(?P<total>\d+)")


class RateTelemetry(BaseModel):
    """Typed view of Salesforce rate-limit signals on a single response."""

    model_config = ConfigDict(frozen=True)

    daily_used: int | None = None
    daily_total: int | None = None
    retry_after: int | None = None  # seconds, when present

    @field_validator("daily_used", "daily_total", "retry_after")
    @classmethod
    def _non_negative(cls, v: int | None) -> int | None:
        if v is not None and v < 0:
            raise ValueError("rate telemetry values must be non-negative")
        return v

    @property
    def daily_remaining(self) -> int | None:
        if self.daily_used is None or self.daily_total is None:
            return None
        return self.daily_total - self.daily_used


def parse_rate_telemetry(headers: CaseInsensitiveDict[str]) -> RateTelemetry:
    """Extract limit telemetry from Salesforce response headers.

    Missing or malformed headers degrade to None rather than raising, so a
    throttled response never crashes the parser.
    """
    used = total = None
    match = _USAGE_RE.search(headers.get("Sforce-Limit-Info", ""))
    if match:
        used, total = int(match["used"]), int(match["total"])

    retry_after = None
    raw = headers.get("Retry-After", "").strip()
    if raw.isdigit():
        retry_after = int(raw)

    return RateTelemetry(daily_used=used, daily_total=total, retry_after=retry_after)

Parsing telemetry upfront lets the connector throttle itself before it exhausts the allocation, rather than reacting only after a 429.

Step 2 — Deterministic backoff with cryptographic jitter #

Plain exponential backoff synchronizes distributed workers into a thundering herd that re-triggers Salesforce’s adaptive throttling. Add jitter from the secrets module so no two workers wake on the same tick, and cap the total delay far below the compliance SLA. When Salesforce does supply Retry-After, honor it as a floor.

import secrets
import time
from collections.abc import Callable
from functools import wraps

from requests import Response
from requests.exceptions import ConnectionError as ReqConnectionError, HTTPError

MAX_RETRIES = 5
BASE_DELAY = 2.0            # seconds
BACKOFF_CEILING = 300.0    # per-call cap, << statutory SLA


def deterministic_retry(func: Callable[..., Response]) -> Callable[..., Response]:
    """Retry only on 429/connection errors with jittered exponential backoff."""

    @wraps(func)
    def wrapper(*args: object, **kwargs: object) -> Response:
        for attempt in range(MAX_RETRIES):
            try:
                response = func(*args, **kwargs)
                response.raise_for_status()
                return response
            except HTTPError as exc:
                if exc.response is None or exc.response.status_code != 429:
                    raise
                telemetry = parse_rate_telemetry(exc.response.headers)
                jitter = secrets.randbelow(1000) / 1000.0
                delay = min(BASE_DELAY * (2 ** attempt) + jitter, BACKOFF_CEILING)
                if telemetry.retry_after is not None:
                    delay = max(delay, float(telemetry.retry_after))
                time.sleep(delay)
            except ReqConnectionError:
                time.sleep(min(BASE_DELAY * (2 ** attempt), BACKOFF_CEILING))
        raise RuntimeError("Salesforce retries exhausted for DSR extraction call")

    return wrapper

secrets.randbelow yields non-predictable jitter, which desynchronizes concurrent nodes more reliably than a seeded PRNG and avoids reinforcing platform-side adaptive throttling. This backoff discipline is the Salesforce-specific counterpart to the retry semantics used in Implementing Celery for Async Polling, where the same DSR SLA constraint bounds the worker retry budget.

Step 3 — Buffer extracted PII to encrypted disk before sleeping #

During a multi-minute backoff, extracted personal data should not linger in a worker’s heap where a crash dump or swap can expose it. Serialize each record to an encrypted, owner-only file before the connector transitions into a wait or DLQ state, satisfying the security-of-processing duty in GDPR Article 32.

import json
import os
from pathlib import Path

from cryptography.fernet import Fernet


class SecureTransientBuffer:
    """Encrypts extracted PII to owner-only disk during backoff windows."""

    def __init__(self, encryption_key: bytes, buffer_dir: str = "/var/lib/dsr/buffer"):
        self._cipher = Fernet(encryption_key)  # rotate via a KMS-backed key
        self._dir = Path(buffer_dir)
        self._dir.mkdir(parents=True, exist_ok=True, mode=0o700)

    def stash(self, record_id: str, payload: dict) -> Path:
        """Encrypt and persist a record; returns the buffer path."""
        blob = self._cipher.encrypt(json.dumps(payload, separators=(",", ":")).encode())
        path = self._dir / f"{record_id}.enc"
        # Write via temp + atomic rename so no partial plaintext-length file lingers.
        tmp = path.with_suffix(".tmp")
        tmp.write_bytes(blob)
        os.chmod(tmp, 0o600)
        tmp.replace(path)
        return path

    def load(self, path: Path) -> dict:
        return json.loads(self._cipher.decrypt(path.read_bytes()).decode())

    def purge(self, path: Path) -> None:
        """Remove the buffer once the record is safely committed downstream."""
        path.unlink(missing_ok=True)

The Fernet key must come from a managed key store, not Fernet.generate_key() at runtime — an ephemeral key that dies with the worker makes every buffered record unrecoverable, which is its own compliance failure.

Step 4 — Enforce idempotency on writes with conditional requests #

If a DSR update (for example, an erasure flag) succeeds but the acknowledgement drops, a blind retry duplicates the mutation. Salesforce honors If-Match ETags, so a conditional PATCH re-executes safely and returns 412 Precondition Failed if the record changed under you.

from requests import Response, Session


def idempotent_patch(
    session: Session, endpoint: str, payload: dict, etag: str | None = None
) -> tuple[Response, str | None]:
    """Conditionally update a record; a dropped ack cannot double-apply."""
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if etag:
        headers["If-Match"] = etag  # write only if the record is unchanged
    response = session.patch(endpoint, json=payload, headers=headers)
    return response, response.headers.get("ETag")

On 412, re-fetch the record, reconcile, and only then re-issue the write — never retry the mutation blindly.

Step 5 — Route exhausted requests to a dead-letter queue #

When the retry ceiling is hit or the daily allocation is depleted, degrade gracefully: envelope the request and route it to a monitored DLQ for adjudication once limits reset. The in-memory queue below must be backed by durable storage in production; the same DLQ discipline governs the Async Polling & Queue Management layer.

import queue
import time
from dataclasses import dataclass, field


@dataclass(frozen=True)
class DSRRequestEnvelope:
    record_id: str
    buffer_path: str          # points at the encrypted payload, not the PII
    retry_count: int
    failure_reason: str
    enqueued_at: float = field(default_factory=time.time)


class DLQRouter:
    """Compliance safety valve for requests that cannot complete in-window."""

    def __init__(self, max_size: int = 10_000):
        self._queue: queue.Queue[DSRRequestEnvelope] = queue.Queue(maxsize=max_size)

    def route(self, envelope: DSRRequestEnvelope) -> bool:
        try:
            self._queue.put_nowait(envelope)
            return True
        except queue.Full:
            self._persist(envelope)  # spill to durable store when memory saturates
            return False

    def _persist(self, envelope: DSRRequestEnvelope) -> None:
        # Append to an encrypted, append-only store (e.g. S3 with SSE, or a WORM ledger).
        raise NotImplementedError

The envelope carries the buffer path, never the plaintext, so the DLQ record itself contains no personal data. Operators monitor queue depth, prioritize by DSR deadline proximity, and reprocess once the allocation window resets.

Configuration reference #

Verification #

Confirm the backoff never exceeds the per-call ceiling and that a 429 triggers exactly one retry cycle before succeeding. A responses-mocked unit test isolates the connector from live Salesforce:

import responses
from requests import Session


@responses.activate
def test_backoff_recovers_after_single_429(monkeypatch):
    sleeps: list[float] = []
    monkeypatch.setattr("time.sleep", sleeps.append)  # capture, don't wait

    url = "https://example.my.salesforce.com/services/data/v60.0/query"
    responses.add(responses.GET, url, status=429,
                  headers={"Sforce-Limit-Info": "api-usage=4999999/5000000"})
    responses.add(responses.GET, url, status=200, json={"records": []})

    session = Session()

    @deterministic_retry
    def fetch() -> object:
        return session.get(url)

    resp = fetch()
    assert resp.status_code == 200
    assert len(sleeps) == 1                 # exactly one backoff
    assert 2.0 <= sleeps[0] <= BACKOFF_CEILING

In production, assert on structured logs: every throttled call should emit one line tagged THROTTLED carrying daily_remaining, the chosen delay, and the retry attempt — this is the auditable evidence that throttling was handled inside the SLA, feeding the batch-level telemetry the SaaS API Sync Strategies layer reports.

Troubleshooting #

Retries never fire, calls fail immediately : raise_for_status() is being called somewhere before the decorator sees the response, or the error is not a 429. The decorator intentionally re-raises non-429 HTTP errors (auth drift, validation) so they surface fast instead of burning the retry budget.

Backoff grows unbounded / breaches the SLA : BACKOFF_CEILING is unset or too high, or Retry-After from Salesforce exceeds the ceiling and is being honored blindly. Cap the honored Retry-After and, if it alone would breach the window, dead-letter immediately rather than sleep.

Duplicate erasure or update after a dropped acknowledgement : The mutation was retried without an If-Match ETag. Capture the ETag from the prior read and attach it to every PATCH; on 412, re-fetch and reconcile before rewriting.

Buffered records cannot be decrypted after a restart : The Fernet key was generated at runtime and died with the worker. Source the key from a KMS or secrets manager and version it so buffers written under a rotated key remain readable.

Daily allocation depletes mid-batch : The connector is not reading daily_remaining proactively. Gate new batch dispatch when daily_remaining falls below a reserve threshold and route the remainder to the DLQ for the next allocation window instead of forcing 429s.

Related #

• SaaS API Sync Strategies — parent guide to credential isolation, cursor normalization, and SLA telemetry across heterogeneous SaaS APIs.
• Implementing Celery for Async Polling — the distributed worker layer that carries these retry and DLQ semantics across all connectors.
• 30-Day vs 45-Day SLA Mapping — how the GDPR one-month and CCPA 45-day deadlines translate into the retry ceilings this connector must respect.
• Cross-System Data Discovery & Sync — the pipeline stage this connector feeds with a deterministic, audit-defensible record manifest.