Patrón Claim Check

Descripción General

El Patrón Claim Check almacena un payload grande en almacenamiento externo y envía solo un token de referencia ligero a través de la infraestructura de mensajería. El consumidor usa ese token para obtener el payload por su cuenta, así que el broker nunca transporta los datos pesados.

Los brokers imponen límites de tamaño por una razón: SQS limita los mensajes a 256KB y Kafka usa por defecto alrededor de 1MB. Un reporte generado, un archivo de video o una exportación masiva pueden superar esos límites, y subir el límite suele cambiar un problema por otro peor.

Cuándo Usar

Usa el Patrón Claim Check cuando:

• Los payloads exceden el límite de tamaño del broker y subirlo no es una opción — servicios gestionados como SQS no lo permiten.
• Los mensajes grandes ralentizan el broker o consumen demasiada memoria y disco.
• Varios consumidores necesitan leer el mismo payload grande; cada uno puede descargarlo del storage en lugar de recibir una copia por el bus.
• Los productores generan datos más rápido de lo que el pipeline los drena y no quieres que un payload enorme bloquee todos los mensajes detrás de él.

Cuándo Evitar

Evítalo cuando:

• Los payloads caben cómodamente dentro de los límites del broker — el salto extra al storage solo añade latencia y un punto de fallo.
• El coste del almacenamiento externo (peticiones, egreso, limpieza de retención) supera lo que ahorras en el broker.
• El consumidor necesita el payload rápido; el claim check añade un segundo ciclo de red por mensaje.
• No hay un almacenamiento de objetos fiable (S3, Azure Blob, GCS) o el payload y el mensaje deben confirmarse atómicamente en una transacción.

Por Qué los Brokers Sufren con Mensajes Grandes

Los brokers de mensajes están hechos para enrutar sobres pequeños rápido, no para mover archivos. Un payload de 20MB duele de varias formas a la vez: ocupa memoria del broker o su log mientras cada consumidor se pone al día, retrasa todo lo que está en cola detrás (bloqueo de cabeza de línea), y en brokers replicados como Kafka cada byte se copia a los seguidores, así que un mensaje grande multiplica el coste de almacenamiento y red en todo el clúster. Los productores también lo pagan — una publicación que tarda segundos es fuente natural de timeouts, reintentos y duplicados aguas abajo.

Los límites de tamaño existen precisamente para proteger al broker de esto. SQS rechaza cualquier cosa por encima de 256KB, Kafka rechaza mensajes por encima de message.max.bytes (1MB por defecto), e incluso brokers que aceptan payloads mayores, como RabbitMQ con 16MB, se degradan mucho antes del límite. Cuando el payload es más grande que el sobre, el sobre debería llevar un puntero en su lugar. La guía de colas de mensajería compara cómo RabbitMQ, Kafka y SQS se degradan al crecer los payloads.

Solución

Python

import uuid
import json
from typing import Optional
from datetime import datetime

# ============================================================================
# ALMACENAMIENTO EXTERNO (simulando S3 / Azure Blob / GCS)
# ============================================================================

class BlobStorage:
    """Almacén de blobs en memoria para demostración"""
    def __init__(self):
        self._store = {}

    def upload(self, data: bytes, content_type: str = "application/json") -> str:
        token = str(uuid.uuid4())
        self._store[token] = {
            "data": data,
            "content_type": content_type,
            "created_at": datetime.now()
        }
        return token

    def download(self, token: str) -> Optional[bytes]:
        entry = self._store.get(token)
        return entry["data"] if entry else None

    def delete(self, token: str):
        self._store.pop(token, None)

# ============================================================================
# BROKER DE MENSAJES (simulando RabbitMQ / Kafka / SQS)
# ============================================================================

class MessageBroker:
    def __init__(self):
        self._queues = {}

    def publish(self, queue: str, message: dict):
        self._queues.setdefault(queue, []).append(message)

    def consume(self, queue: str) -> Optional[dict]:
        messages = self._queues.get(queue, [])
        return messages.pop(0) if messages else None

# ============================================================================
# IMPLEMENTACIÓN DE CLAIM CHECK
# ============================================================================

class ClaimCheckProducer:
    """Publica mensajes con tokens de claim check en lugar de payloads grandes"""
    def __init__(self, broker: MessageBroker, storage: BlobStorage):
        self.broker = broker
        self.storage = storage

    def publish_large_message(self, queue: str, payload: dict, metadata: Optional[dict] = None):
        # Almacenar payload en storage externo
        payload_bytes = json.dumps(payload).encode("utf-8")
        token = self.storage.upload(payload_bytes, "application/json")

        # Publicar mensaje ligero con referencia de claim check
        message = {
            "claim_check": token,
            "metadata": metadata or {},
            "payload_size": len(payload_bytes),
            "timestamp": datetime.now().isoformat()
        }
        self.broker.publish(queue, message)
        print(f"Publicado claim check: {token} ({len(payload_bytes)} bytes)")
        return token

class ClaimCheckConsumer:
    """Consume mensajes y recupera payloads completos via claim check"""
    def __init__(self, broker: MessageBroker, storage: BlobStorage):
        self.broker = broker
        self.storage = storage

    def process_next(self, queue: str):
        message = self.broker.consume(queue)
        if not message:
            return None

        token = message["claim_check"]
        payload_bytes = self.storage.download(token)

        if payload_bytes is None:
            print(f"ERROR: Claim check {token} no encontrado en storage")
            return None

        payload = json.loads(payload_bytes.decode("utf-8"))

        # Procesar el payload completo
        print(f"Payload recuperado ({message['payload_size']} bytes): {payload['report_id']}")

        # Limpiar payload almacenado después de procesar
        self.storage.delete(token)
        print(f"Claim check eliminado: {token}")

        return payload


# ============================================================================
# USO
# ============================================================================

storage = BlobStorage()
broker = MessageBroker()

producer = ClaimCheckProducer(broker, storage)
consumer = ClaimCheckConsumer(broker, storage)

# Payload grande: un reporte detallado que excedería los límites típicos del broker
large_report = {
    "report_id": "RPT-2024-Q1",
    "generated_at": datetime.now().isoformat(),
    "records": [
        {"id": i, "data": "x" * 1000} for i in range(1000)  # Payload de 1MB+
    ],
    "summary": {"total": 1000, "revenue": 500000.00}
}

producer.publish_large_message("reports.queue", large_report, {"priority": "high"})
consumer.process_next("reports.queue")

Java

import java.util.*;
import java.util.concurrent.*;

// Abstracción de blob storage
interface BlobStorage {
    String upload(byte[] data, String contentType);
    byte[] download(String token);
    void delete(String token);
}

class InMemoryBlobStorage implements BlobStorage {
    private final Map<String, byte[]> store = new ConcurrentHashMap<>();

    public String upload(byte[] data, String contentType) {
        String token = UUID.randomUUID().toString();
        store.put(token, data);
        return token;
    }

    public byte[] download(String token) {
        return store.get(token);
    }

    public void delete(String token) {
        store.remove(token);
    }
}

// Message broker
class MessageBroker {
    private final Map<String, Queue<Map<String, Object>>> queues = new ConcurrentHashMap<>();

    public void publish(String queue, Map<String, Object> message) {
        queues.computeIfAbsent(queue, k -> new ConcurrentLinkedQueue<>()).offer(message);
    }

    public Map<String, Object> consume(String queue) {
        Queue<Map<String, Object>> q = queues.get(queue);
        return q != null ? q.poll() : null;
    }
}

// Claim Check Producer
class ClaimCheckProducer {
    private final MessageBroker broker;
    private final BlobStorage storage;

    public ClaimCheckProducer(MessageBroker broker, BlobStorage storage) {
        this.broker = broker; this.storage = storage;
    }

    public String publishLargeMessage(String queue, String payloadJson, Map<String, Object> metadata) {
        byte[] data = payloadJson.getBytes();
        String token = storage.upload(data, "application/json");

        Map<String, Object> message = new HashMap<>();
        message.put("claimCheck", token);
        message.put("metadata", metadata);
        message.put("payloadSize", data.length);
        message.put("timestamp", new Date().toInstant().toString());

        broker.publish(queue, message);
        System.out.println("Publicado claim check: " + token + " (" + data.length + " bytes)");
        return token;
    }
}

// Claim Check Consumer
class ClaimCheckConsumer {
    private final MessageBroker broker;
    private final BlobStorage storage;

    public ClaimCheckConsumer(MessageBroker broker, BlobStorage storage) {
        this.broker = broker; this.storage = storage;
    }

    public String processNext(String queue) {
        Map<String, Object> message = broker.consume(queue);
        if (message == null) return null;

        String token = (String) message.get("claimCheck");
        byte[] data = storage.download(token);

        if (data == null) {
            System.err.println("ERROR: Claim check no encontrado: " + token);
            return null;
        }

        String payload = new String(data);
        System.out.println("Payload recuperado (" + message.get("payloadSize") + " bytes)");
        storage.delete(token);
        System.out.println("Claim check eliminado: " + token);
        return payload;
    }
}

// Uso
BlobStorage storage = new InMemoryBlobStorage();
MessageBroker broker = new MessageBroker();
ClaimCheckProducer producer = new ClaimCheckProducer(broker, storage);
ClaimCheckConsumer consumer = new ClaimCheckConsumer(broker, storage);

String largePayload = "{\"report_id\":\"RPT-001\",\"data\":\"" + "x".repeat(10000) + "\"}";
producer.publishLargeMessage("reports", largePayload, Map.of("priority", "high"));
consumer.processNext("reports");

JavaScript

const crypto = require('crypto');

// Blob storage
class InMemoryBlobStorage {
  constructor() {
    this.store = new Map();
  }

  upload(data, contentType) {
    const token = crypto.randomUUID();
    this.store.set(token, { data, contentType, createdAt: new Date() });
    return token;
  }

  download(token) {
    const entry = this.store.get(token);
    return entry ? entry.data : null;
  }

  delete(token) {
    this.store.delete(token);
  }
}

// Message broker
class MessageBroker {
  constructor() {
    this.queues = new Map();
  }

  publish(queue, message) {
    if (!this.queues.has(queue)) this.queues.set(queue, []);
    this.queues.get(queue).push(message);
  }

  consume(queue) {
    const messages = this.queues.get(queue);
    return messages && messages.length > 0 ? messages.shift() : null;
  }
}

// Claim Check Producer
class ClaimCheckProducer {
  constructor(broker, storage) {
    this.broker = broker;
    this.storage = storage;
  }

  publishLargeMessage(queue, payload, metadata = {}) {
    const payloadBytes = Buffer.from(JSON.stringify(payload));
    const token = this.storage.upload(payloadBytes, 'application/json');

    const message = {
      claimCheck: token,
      metadata,
      payloadSize: payloadBytes.length,
      timestamp: new Date().toISOString()
    };

    this.broker.publish(queue, message);
    console.log(`Publicado claim check: ${token} (${payloadBytes.length} bytes)`);
    return token;
  }
}

// Claim Check Consumer
class ClaimCheckConsumer {
  constructor(broker, storage) {
    this.broker = broker;
    this.storage = storage;
  }

  processNext(queue) {
    const message = this.broker.consume(queue);
    if (!message) return null;

    const payloadBytes = this.storage.download(message.claimCheck);
    if (!payloadBytes) {
      console.error(`ERROR: Claim check no encontrado: ${message.claimCheck}`);
      return null;
    }

    const payload = JSON.parse(payloadBytes.toString());
    console.log(`Payload recuperado (${message.payloadSize} bytes)`);

    this.storage.delete(message.claimCheck);
    console.log(`Claim check eliminado: ${message.claimCheck}`);

    return payload;
  }
}

// Uso
const storage = new InMemoryBlobStorage();
const broker = new MessageBroker();
const producer = new ClaimCheckProducer(broker, storage);
const consumer = new ClaimCheckConsumer(broker, storage);

const largePayload = {
  reportId: 'RPT-001',
  records: Array.from({ length: 1000 }, (_, i) => ({ id: i, data: 'x'.repeat(1000) }))
};

producer.publishLargeMessage('reports', largePayload, { priority: 'high' });
consumer.processNext('reports');

Explicación

El patrón divide cada mensaje en dos canales: el sobre viaja por el broker, los datos viajan por el storage.

1. Almacenar: el productor sube el payload al almacenamiento de objetos y recibe un identificador — una clave, una URL o un token opaco.
2. Referenciar: el productor publica un mensaje pequeño con esa referencia más metadatos (tamaño, content type, checksum).
3. Recuperar: el consumidor lee la referencia y descarga el payload directamente del storage.
4. Limpiar: el objeto almacenado se elimina cuando ya nadie lo necesita.

Como el payload y el mensaje viven en dos sistemas, no puedes actualizar ambos atómicamente. Los fallos clásicos son: un mensaje que llega antes de que el payload sea legible, el mismo mensaje entregado dos veces — la entrega at-least-once es lo habitual — y un payload que ya no existe cuando llega el consumidor. Diseña pensando en los tres: publica el token solo después de que la subida tenga éxito, reintenta las referencias faltantes brevemente antes de rendirte, y haz el procesamiento idempotente combinando este patrón con el Patrón Consumidor Idempotente. Un checksum dentro del mensaje permite al consumidor verificar la descarga antes de confiar en ella.

• Código companion: el repo companion de claim-check-pattern empaqueta estos ejemplos como archivos ejecutables en Python, JavaScript y Java.

Estrategias de Limpieza

La pregunta operativa más difícil es quién elimina el payload, y cuándo. Cuatro estrategias cubren la mayoría de los sistemas:

Estrategia	Cómo funciona	Mejor para	Ojo con
El consumidor elimina	El consumidor borra el objeto justo después de procesar	Un solo consumidor, datos de vida corta	Los payloads quedan huérfanos si el consumidor muere entre procesar y eliminar
TTL / reglas de ciclo de vida	El storage expira objetos tras N horas o días	Cualquier carga de trabajo; cero código	Un TTL demasiado corto deja a los consumidores con referencias muertas
Conteo de referencias	Cada consumidor decrementa un contador; el último elimina	Fan-out a un conjunto conocido de consumidores	El contador necesita actualizaciones atómicas y su propio manejo de fallos
Job barredor	Un worker programado elimina objetos más viejos que un umbral	Red de seguridad junto a otra estrategia	Una pieza móvil más que desplegar y monitorizar

La mayoría de los sistemas en producción combinan dos: el consumidor elimina en el camino feliz y una regla de ciclo de vida (S3 lifecycle, Azure Blob lifecycle management) recoge lo que un crash o un mensaje venenoso dejó atrás. Nunca dependas solo de la eliminación por el consumidor — un consumidor que muere entre procesar y eliminar fuga almacenamiento en silencio.

Seguridad e Integridad

Un claim check funciona como una capability: quien tiene el token puede leer el payload. Trátalo en consecuencia.

• Encripta los payloads en reposo (SSE-S3, SSE-KMS o equivalente) y exige HTTPS para la recuperación.
• Prefiere URLs pre-firmadas de corta duración antes que credenciales del bucket embebidas en el mensaje.
• Haz los tokens impredecibles — uuid4 o claves aleatorias funcionan; los IDs secuenciales permiten enumerar tus payloads.
• Envía un hash SHA-256 del payload en el mensaje para que el consumidor detecte una descarga truncada o intercambiada.
• Acota los permisos de storage: los productores escriben, los consumidores leen y eliminan, todo solo sobre el prefijo de claim-check.

Variantes

La mecánica es la misma en todas partes; lo que cambia es dónde vive el payload y cómo lo alcanzan los consumidores.

Variante	Almacenamiento	Caso de Uso
S3 Claim Check	Amazon S3 / URLs pre-firmadas	Multi-región, duradero y barato
Shared Volume	NFS / EFS / SMB	On-premise, baja latencia, misma red
Database BLOB	PostgreSQL BYTEA / MySQL BLOB	Cuando se necesitan transacciones con metadata
CDN	CloudFront / Cloudflare	Contenido público que los consumidores descargan directamente
Streaming	S3 + peticiones por rango de bytes	Video/audio donde los consumidores buscan posiciones

Elige el storage como lo elegirías para cualquier otro problema de archivos grandes — durabilidad, latencia hacia los consumidores, coste — porque la parte de mensajería no cambia.

Lo que funciona

• Usa URLs pre-firmadas para seguridad. En lugar de exponer el storage directamente, el mensaje lleva una URL de tiempo limitado.
• Expira los payloads automáticamente. Las reglas de ciclo de vida del storage eliminan claim checks viejos sin una línea de código de limpieza.
• Envía un checksum en el mensaje. Un hash del payload permite al consumidor detectar una descarga truncada o intercambiada antes de procesarla.
• Planifica para payloads faltantes. Si un claim check no apunta a nada, reintenta brevemente, luego loguea y envía a dead-letter en lugar de romper el bucle.
• Comprime antes de almacenar. Gzip o Brotli reduce tanto la factura de storage como el tiempo de descarga en payloads de texto grandes.

Errores Comunes

• No limpiar nunca. Sin eliminación ni TTL, los claim checks se acumulan y la factura de storage crece en silencio mes a mes.
• Almacenar datos sensibles sin encriptar. Encripta en reposo y descarga por HTTPS; un token impredecible no es sustituto.
• Emitir URLs que nunca expiran. Una URL pre-firmada filtrada sin expiración es una clave de lectura permanente.
• Asumir que el payload está ahí. Las reglas de ciclo de vida mal configuradas borran objetos antes de tiempo y los stores caseros los pierden — el consumidor debe manejar referencias faltantes.
• Descargar de forma síncrona en el bucle del consumidor. Descarga asíncronamente o delega a un worker para que una descarga lenta de 50MB no detenga toda la cola.

Ejemplos del Mundo Real

AWS SQS + S3

SQS rechaza cualquier cosa por encima de 256KB, y la propia respuesta de AWS es el claim check: la SQS Extended Client Library almacena el payload en S3 y envía el puntero del objeto en el mensaje, manejando payloads de hasta 2GB. Puedes hacer lo mismo con llamadas planas al SDK — la librería solo oculta la fontanería.

Azure Service Bus + Blob Storage

Azure Service Bus acepta hasta 256KB en Standard y hasta 100MB en Premium, pero cada entidad sigue por defecto con un tope de 1MB que hay que subir explícitamente. El centro de arquitectura de Microsoft documenta el claim check como la solución estándar: subir a Blob Storage y enviar el URI del blob en el mensaje.

Kafka + Object Storage

El message.max.bytes por defecto de Kafka ronda 1MB, y subirlo cuesta memoria del broker y ancho de banda de replicación. Los equipos que producen payloads grandes — reportes, medios, exportaciones batch — los suben a S3 o HDFS y mantienen el registro de Kafka por debajo de unos pocos KB.

Lecturas Adicionales

• Patrón Claim Check — Azure Architecture Center: la versión de Microsoft, con el flujo de Service Bus y la guía de seguridad del token.
• Gestión de mensajes grandes de Amazon SQS: la documentación oficial de la Extended Client Library para payloads de hasta 2GB.
• Enterprise Integration Patterns: el catálogo de Hohpe y Woolf, donde el patrón se documentó por primera vez.