Handling CCPA Deletion vs Opt-Out Requests in DSR Pipelines

The single most frequent execution fault in a California-facing Data Subject Request (DSR) pipeline is treating a deletion request and an opt-out request as one workflow. They share an intake channel and an identity-verification gate, but downstream they are opposites: a deletion under CCPA §1798.105 mandates irreversible erasure across primary stores, backups, and service providers, subject to the narrow retention exceptions of §1798.105(d); an opt-out of sale or sharing under §1798.120 mandates the exact opposite — the data must be preserved and merely flagged, with a persistent, revocable suppression state enforced on every downstream flow. This page belongs to the GDPR vs CCPA Request Taxonomies cluster within the broader DSR Architecture & Intake Routing framework, and it addresses the specific gap the taxonomy leaves open: how you split one verified consumer submission into two structurally incompatible execution paths without either purging data you were obliged to keep or retaining data you were obliged to destroy. The engineer who reaches this page has a working intake gate and needs the routing and execution layer that keeps these two obligations from ever touching.

The lifecycle below is what every design choice on this page protects: a verified CCPA request is classified into atomic intents, routed by jurisdictional precedence, and then diverges permanently — deletion destroys data and emits proof of erasure, opt-out preserves data under a versioned suppression flag.

Prerequisites #

• Python 3.11+ — the code uses asyncio.TaskGroup, PEP 604 unions, and zoneinfo-aware UTC timestamps.
• Pydantic v2 (pydantic>=2.6) for boundary validation using ConfigDict and field_validator.
• A message bus (Kafka or a cloud pub/sub) for cross-system suppression-event propagation, plus a versioned central policy/preference store the API gateway can read on every request.
• An append-only (WORM) audit store for routing decisions and proof-of-erasure records, and a dead-letter queue (DLQ) for unroutable payloads.
• Infrastructure: every store DSN, bus endpoint, and secret arrives through environment variables validated at startup — never hardcoded. Deletion handlers additionally need write credentials scoped to each service provider you must notify under §1798.105©.

This page assumes the verification and normalization stages are already in place — the identity-proofing and payload-shaping described in Secure Intake Form Design and the framework-assignment logic of Jurisdiction Routing Logic. The code here replaces ad-hoc if request.type == "delete" branching, not the gate underneath it.

Step-by-step implementation #

Step 1 — Classify intent and split hybrid payloads atomically #

Consumer submissions rarely map cleanly to a single statutory category. A user who asks to “close my account and stop selling my info” has emitted both a §1798.105 deletion and a §1798.120 opt-out, and each must become its own tracked, idempotent action — collapsing them into one record makes the audit trail unreconstructable and risks executing only one half. Validate at the boundary with Pydantic v2, reject empty intent arrays (which otherwise start an SLA clock against nothing), and fan a hybrid payload out into atomic sub-requests that share an idempotency context.

import uuid
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, field_validator


class DSRIntent(str, Enum):
    DELETION = "deletion"                       # CCPA §1798.105
    OPT_OUT_SALE = "opt_out_sale"               # CCPA §1798.120
    OPT_OUT_SHARING = "opt_out_sharing"         # CPRA amendment to §1798.120
    OPT_OUT_SENSITIVE = "opt_out_sensitive"     # limit use of sensitive PI, §1798.121


class DSRPayload(BaseModel):
    """A verified CCPA request at the routing boundary."""

    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    request_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    consumer_id: str
    jurisdiction: str
    intents: list[DSRIntent]
    verified: bool = False
    idempotency_key: str

    @field_validator("intents")
    @classmethod
    def non_empty(cls, v: list[DSRIntent]) -> list[DSRIntent]:
        if not v:
            raise ValueError("empty intent array would start an SLA clock against no action")
        return v

    def split(self) -> list["DSRPayload"]:
        """Fan a hybrid payload into atomic sub-requests sharing an idempotency root."""
        return [
            DSRPayload(
                consumer_id=self.consumer_id,
                jurisdiction=self.jurisdiction,
                intents=[intent],
                verified=self.verified,
                idempotency_key=f"{self.idempotency_key}:{intent.value}",
            )
            for intent in self.intents
        ]

extra="forbid" enforces data minimization at the boundary per CCPA §1798.100©: any undeclared field is rejected rather than silently carried into a regulated workflow.

Step 2 — Route by jurisdictional precedence #

Once split, each atomic request is matched against a precedence matrix before it is queued. Precedence matters because a single consumer can carry overlapping obligations — a California IP address against an account governed by an EU processing agreement — and the pipeline must resolve to the stricter deadline rather than the first rule that happens to match. Deletion (§1798.105) is queued to an erasure worker; any opt-out variant (§1798.120 / §1798.121) is queued to a suppression worker with a distinct handler. The 45-day CCPA response baseline of §1798.130(a)(2) and the tighter one-month GDPR window of Article 12(3) are carried on the rule itself so the SLA timer is set at routing time; the deadline arithmetic is expanded in 30-Day vs 45-Day SLA Mapping.

from collections.abc import Callable
from dataclasses import dataclass


@dataclass(frozen=True)
class RoutingRule:
    condition: Callable[[DSRPayload], bool]
    target_queue: str
    sla_days: int
    precedence: int  # lower wins


JURISDICTION_MATRIX: dict[str, RoutingRule] = {
    "CA_DELETION": RoutingRule(
        condition=lambda p: p.jurisdiction == "CA" and DSRIntent.DELETION in p.intents,
        target_queue="ccpa.erasure",
        sla_days=45,   # CCPA §1798.130(a)(2)
        precedence=2,
    ),
    "CA_OPT_OUT": RoutingRule(
        condition=lambda p: p.jurisdiction == "CA"
        and any(i is not DSRIntent.DELETION for i in p.intents),
        target_queue="ccpa.suppression",
        sla_days=45,   # opt-out honored "as soon as feasibly possible", §1798.135
        precedence=3,
    ),
    "EU_ERASURE": RoutingRule(
        condition=lambda p: p.jurisdiction == "EU" and DSRIntent.DELETION in p.intents,
        target_queue="gdpr.erasure",
        sla_days=30,   # GDPR Art. 12(3) — stricter, so lower precedence value
        precedence=1,
    ),
}


def resolve_routing(payload: DSRPayload) -> RoutingRule:
    """Return the highest-precedence (strictest) matching rule, or escalate."""
    matches = [r for r in JURISDICTION_MATRIX.values() if r.condition(payload)]
    if not matches:
        raise RuntimeError("no routing rule matched; escalate to compliance review")
    return min(matches, key=lambda r: r.precedence)

Step 3 — Execute deletion as transactional erasure with a deferred backup tombstone #

Deletion must destroy data and prove it. Purge every primary store transactionally, notify each service provider that received the data (§1798.105©), then write a tombstone that defers backup reconciliation: live snapshots cannot be surgically edited, so a dated tombstone lets the next retention cycle purge the record without breaking backup integrity. Logical deletion alone is insufficient in regulated environments — where overwrite is impractical, apply cryptographic erasure by destroying the record’s encryption key, per the sanitization guidance of NIST SP 800-88 Rev. 1 and the key-lifecycle controls of NIST SP 800-57 Part 1 Rev. 5. Every step appends to the WORM ledger so the proof-of-erasure survives a regulator’s inspection.

import asyncio
from datetime import datetime, timedelta, timezone


async def execute_deletion(
    request_id: str, consumer_id: str, systems: list[str]
) -> dict:
    """Transactional erasure with provider notification and a deferred backup tombstone."""
    audit = {"request_id": request_id, "status": "pending", "systems": []}
    try:
        async with asyncio.TaskGroup() as tg:
            for system in systems:
                tg.create_task(_purge_primary_store(system, consumer_id, request_id))
        # Backups cannot be edited in place: defer purge to the next retention cycle.
        purge_after = datetime.now(timezone.utc) + timedelta(days=90)
        await _create_tombstone(consumer_id, request_id, purge_after)
        await _append_proof_of_erasure(request_id, consumer_id, systems, purge_after)
        audit["status"] = "completed"
        audit["systems"] = systems
    except* Exception as eg:  # TaskGroup raises an ExceptionGroup
        audit["status"] = "failed"
        audit["error"] = "; ".join(str(e) for e in eg.exceptions)
        raise
    return audit


async def _purge_primary_store(system: str, consumer_id: str, request_id: str) -> None:
    """Erase (or crypto-shred) the subject's records in one store; provider-specific SDK."""
    ...


async def _create_tombstone(consumer_id: str, request_id: str, purge_after: datetime) -> None:
    """Record a dated deferred-purge marker so backup snapshots reconcile on schedule."""
    ...


async def _append_proof_of_erasure(
    request_id: str, consumer_id: str, systems: list[str], purge_after: datetime
) -> None:
    """Append an immutable erasure record to the WORM audit ledger."""
    ...

except* is required here: asyncio.TaskGroup collects concurrent failures into an ExceptionGroup, and catching plain Exception would miss them and mark a partial purge as complete.

Step 4 — Execute opt-out as persistent, revocable suppression #

Opt-out is the mirror image: it must not erase anything. Publish an immutable suppression event to the cross-system bus, then upsert a versioned record in the central policy store the API gateway consults before any sale or share. Versioning is what makes the state revocable — §1798.135(a) requires that a consumer be able to opt back in — so you record state transitions rather than overwriting a boolean. The revocable flag and the effective timestamp are the fields downstream enforcement and later reconciliation both depend on.

from datetime import datetime, timezone


async def execute_opt_out(
    request_id: str, consumer_id: str, opt_out_types: list[str]
) -> dict:
    """Publish a suppression event and upsert a versioned, revocable policy record."""
    record = {
        "consumer_id": consumer_id,
        "request_id": request_id,
        "opt_out_types": opt_out_types,
        "effective_date": datetime.now(timezone.utc).isoformat(),
        "revocable": True,           # §1798.135(a) opt-in path must remain open
    }
    await _publish_suppression_event(record)   # immutable, cross-system propagation
    await _upsert_policy_store(consumer_id, record)  # versioned, gateway-readable
    return {"status": "enforced", "record": record}


async def _publish_suppression_event(record: dict) -> None:
    """Emit an immutable suppression event to the cross-system bus (Kafka / pub-sub)."""
    ...


async def _upsert_policy_store(consumer_id: str, record: dict) -> None:
    """Append a new version of the suppression state; never overwrite prior versions."""
    ...

Step 5 — Catch failures with a DLQ and SLA-aware escalation #

Automated pipelines hit transient faults, schema drift, and provider rate limits. Wrap external processor calls in a circuit breaker; serialize unroutable payloads (malformed identity proofs, conflicting jurisdiction claims) to a DLQ with full context; and escalate on the SLA clock, not just on error count. Trigger a P1 escalation once the remaining SLA window drops below a hard threshold so a stuck request cannot silently burn the §1798.130 deadline.

from datetime import datetime, timezone


class DSRFallbackRouter:
    """DLQ persistence plus SLA-threshold escalation for failed DSR executions."""

    def __init__(self, dlq_client, alert_client) -> None:
        self.dlq = dlq_client
        self.alert = alert_client

    async def handle_failure(
        self, request: dict, error: Exception, sla_remaining_hours: float
    ) -> None:
        if sla_remaining_hours < 12:
            await self.alert.escalate(
                request_id=request["request_id"],
                message="Critical SLA threshold breached; manual intervention required.",
                priority="P1",
            )
        await self.dlq.push(
            {
                "original_payload": request,
                "error": str(error),
                "timestamp_utc": datetime.now(timezone.utc).isoformat(),
                "retry_count": 0,
            }
        )

Configuration reference #

The invariant to enforce in review: an opt-out path may never call an erasure primitive, and a deletion path may never write a revocable suppression flag — the two handlers share no code.

Verification #

Confirm the split before trusting the pipeline with real consumer data. Assert that hybrid payloads fan out into atomic intents, that precedence resolves to the strictest rule, and that deletion and opt-out reach disjoint queues.

def test_hybrid_payload_splits_into_atomic_intents() -> None:
    """A deletion + opt-out submission must become two independently keyed requests."""
    p = DSRPayload(
        consumer_id="c-1",
        jurisdiction="CA",
        intents=[DSRIntent.DELETION, DSRIntent.OPT_OUT_SALE],
        idempotency_key="root",
    )
    parts = p.split()
    assert [len(x.intents) for x in parts] == [1, 1]
    assert {x.idempotency_key for x in parts} == {"root:deletion", "root:opt_out_sale"}


def test_precedence_resolves_to_strictest_deadline() -> None:
    """An EU erasure must outrank a CA rule; the 30-day window wins."""
    p = DSRPayload(
        consumer_id="c-2", jurisdiction="EU",
        intents=[DSRIntent.DELETION], idempotency_key="k",
    )
    assert resolve_routing(p).sla_days == 30


def test_empty_intents_rejected() -> None:
    """An empty intent array must never start an SLA clock."""
    import pytest
    with pytest.raises(ValueError):
        DSRPayload(consumer_id="c", jurisdiction="CA", intents=[], idempotency_key="k")

The compliance assertion to hold under load: every completed deletion has exactly one proof-of-erasure ledger entry and a dated tombstone, while every enforced opt-out has exactly one versioned policy record and zero ledger erasure entries. No request should ever appear in both the erasure ledger and the suppression store.

Troubleshooting #

A deletion “succeeds” but records reappear from a backup restore Root cause: no tombstone was written, so a snapshot restore reintroduces purged data. Fix: always write a dated tombstone in _create_tombstone and reconcile every restore against the tombstone table before the data goes live.

An opt-out silently purged the consumer’s data Root cause: a shared handler branched into an erasure primitive, or a hybrid payload was routed as a single “delete + suppress” action. Fix: keep the erasure and suppression handlers code-disjoint and split hybrid payloads in Step 1 so each intent executes independently.

Partial deletion marked as complete Root cause: catching plain Exception around an asyncio.TaskGroup, which raises an ExceptionGroup. Fix: use except* so concurrent per-store failures are observed and the audit status is set to failed.

Requests stall in the queue and breach the 45-day deadline Root cause: escalation keyed on error count, not on the SLA clock, so a request that merely waits never trips an alert. Fix: escalate on sla_remaining_hours as in Step 5 and set the timer from the routing rule’s sla_days.

Cross-boundary conflict routes to the wrong deadline Root cause: routing returns the first matching rule instead of the strictest. Fix: select with min(..., key=precedence) so an EU 30-day erasure outranks a CCPA 45-day rule for the same consumer.

Related #

• GDPR vs CCPA Request Taxonomies — the parent cluster: how overlapping erasure and opt-out concepts are normalized across frameworks.
• How to Map DSR Types to GDPR Article 15 — the sibling access-request mapping this deletion/opt-out split mirrors.
• 30-Day vs 45-Day SLA Mapping — statutory deadline translation for the timers set at routing time.
• Jurisdiction Routing Logic — the framework-assignment stage the precedence matrix depends on.
• DSR Architecture & Intake Routing — the intake and routing control plane this execution layer belongs to.