Property Digest Background Job Queue — Malformed 256KB JSON Schema Validation

Short answer: keep the background job queue for a weekly property-management digest, but make each message a small, versioned reference, reject malformed or larger-than-256KB JSON at the producer, repeat schema validation at the consumer, and send permanently invalid payloads to a dead-letter queue.

The governing trade-off is latency versus cost. Embedding a complete customer data set may avoid one storage read, yet it turns normal growth in leases, maintenance notes, and recipient settings into a hard queue-admission problem. Referencing private data adds a read during processing, but it keeps queue traffic bounded and lets a weekly batch absorb latency away from an interactive request. The queue is the dispatch layer, not the data layer.

That distinction matters more than broker branding.

Define the admission contract before selecting a queue

A weekly digest has at least two clocks. The first measures how long it takes to admit work; the second measures when the completed digest must be ready. Admission should remain short and predictable, while completion can tolerate the extra read needed to resolve a private object ID or URL. If the business instead requires an immediate response after every property update, that extra read belongs in the latency budget and may change the design.

Start with a job ledger keyed by a deterministic digest ID, such as a customer identifier plus the digest period. Store the potentially large input outside the queue and publish an envelope containing the job ID, customer ID, schema version, and private input reference. The worker claims the job ID before doing side effects. A standard queue is at-least-once, so duplicate delivery is expected and consumer idempotency isn't optional.

This placement also separates four limits that teams often blur together. The final UTF-8 JSON bytes must stay within 256KB. A delayed message cannot be delayed more than seven days. Queue retention is at most 30 days, and acknowledgment deletes the message. None of those properties provides Kafka-style replay or independent consumer groups; the durable job ledger has to carry the audit and completion state that the queue does not retain. The awkward part — easy to miss during a schema review — is that a perfectly valid in-memory object can cross the service limit only after JSON escaping and UTF-8 encoding, so admission has to inspect the bytes that will actually travel rather than a character count or a rough estimate of object size. Don't aim at exactly 256KB. The evidence fixes the service ceiling, but it does not establish one universally correct safety margin for every serializer and wrapper. I'm not sure a fixed percentage is useful across clients; measure the exact encoded body produced in the deployed path, choose a lower internal threshold, and test that threshold with long identifiers, escaped characters, and multibyte property names.

The latency consequence is concrete: the worker pays for a private-data lookup. The cost consequence is concrete too: large working sets are stored once instead of being pushed repeatedly through a dispatch channel. No unsupported percentage or benchmark is needed to make that architectural call.

How should a background job queue reject malformed 256KB JSON payloads?

They should enforce the same versioned contract at both boundaries, while assigning failures to permanent or transient classes. Producer validation prevents known-bad work from entering the system. Consumer validation protects the worker from older producers, other publishers, and messages already in flight during a deployment.

Order matters. Serialize first, count the resulting UTF-8 bytes, then publish. On consume, check the byte count, decode UTF-8, parse JSON, validate the object shape, and only then claim the job ID. A syntax error such as malformed JSON cannot become valid on a later delivery. Neither can a missing required field, an unsupported schema version, or an oversized body. Those payloads belong in the DLQ rather than an indefinite retry cycle.

The following Python program is intentionally broker-independent. It defines the application envelope rather than inventing a vendor request wrapper, and it produces stable reason codes that a Node.js producer and consumer can implement with their normal schema library. Run it as written to exercise a valid message and a malformed one.

import json
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Any


MAX_MESSAGE_BYTES = 256 * 1024
SUPPORTED_SCHEMA_VERSIONS = {1}


class PermanentPayloadError(ValueError):
    def __init__(self, code: str, detail: str) -> None:
        super().__init__(f"{code}: {detail}")
        self.code = code


@dataclass(frozen=True)
class DigestMessage:
    schema_version: int
    job_id: str
    customer_id: str
    digest_input_ref: str


def validate_shape(value: Any) -> DigestMessage:
    if not isinstance(value, dict):
        raise PermanentPayloadError("SCHEMA_OBJECT", "payload must be an object")

    fields = {
        "schema_version": int,
        "job_id": str,
        "customer_id": str,
        "digest_input_ref": str,
    }
    for name, expected_type in fields.items():
        if name not in value or type(value[name]) is not expected_type:
            raise PermanentPayloadError(
                "SCHEMA_FIELD",
                f"{name} must be {expected_type.__name__}",
            )

    if value["schema_version"] not in SUPPORTED_SCHEMA_VERSIONS:
        raise PermanentPayloadError(
            "SCHEMA_VERSION",
            f"unsupported version {value['schema_version']}",
        )

    return DigestMessage(**{name: value[name] for name in fields})


def encode_for_publish(value: Any) -> bytes:
    message = validate_shape(value)
    body = json.dumps(
        message.__dict__,
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    if len(body) > MAX_MESSAGE_BYTES:
        raise PermanentPayloadError(
            "PAYLOAD_TOO_LARGE",
            f"encoded payload is {len(body)} bytes",
        )
    return body


def decode_after_consume(body: bytes) -> DigestMessage:
    if len(body) > MAX_MESSAGE_BYTES:
        raise PermanentPayloadError(
            "PAYLOAD_TOO_LARGE",
            f"received payload is {len(body)} bytes",
        )
    try:
        value = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise PermanentPayloadError("MALFORMED_JSON", str(error)) from error
    return validate_shape(value)


def publish(value: Any, attempts: int = 4) -> dict[str, Any]:
    body = encode_for_publish(value)
    base_url = os.environ["INFRAI_BASE_URL"].rstrip("/")
    request = urllib.request.Request(
        f"{base_url}/queue/publish",
        data=body,
        method="POST",
        headers={
            "Authorization": f"Bearer {os.environ['INFRAI_API_KEY']}",
            "Content-Type": "application/json",
            "Idempotency-Key": value["job_id"],
        },
    )

    for attempt in range(attempts):
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                return json.load(response)
        except urllib.error.HTTPError as error:
            response_body = error.read().decode("utf-8", errors="replace")
            if error.code != 429 or attempt == attempts - 1:
                raise RuntimeError(
                    f"publish returned HTTP {error.code}: {response_body}"
                ) from error
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)

    raise RuntimeError("publish attempts exhausted")


if __name__ == "__main__":
    outgoing = {
        "schema_version": 1,
        "job_id": "digest-2026-w32-customer-1842",
        "customer_id": "customer-1842",
        "digest_input_ref": "private-object-id-7f31",
    }
    encoded = encode_for_publish(outgoing)
    print(decode_after_consume(encoded))

    try:
        decode_after_consume(b'{"schema_version":1,')
    except PermanentPayloadError as error:
        print(error.code)

    if os.environ.get("INFRAI_API_KEY"):
        print(publish(outgoing))

Versioning does not mean accepting any version. A consumer should declare the versions it understands; an unknown version is a terminal contract error for that deployment. New optional fields can be introduced only under a compatibility policy that old consumers can tolerate. A breaking shape needs a new version and a rollout in which compatible consumers exist before that version is published.

Be equally strict about error ownership. A temporary failure while reading the referenced private object may justify bounded retry. A parsing error does not. Mixing those cases makes one poison message consume worker capacity again and again, even though time cannot change its bytes.

Compare recovery models after fixing the contract

Once the envelope is small and validated, compare systems by the recovery behavior the digest actually needs. This isn't a universal leaderboard.

Option	Fit for the weekly digest	Latency and cost implication	Not suitable when
Infrai queue	Its plain REST API needs no SDK or client-library version, and one key can cover scheduling and queue calls; the verified write/read paths are POST /v1/queue/publish and POST /v1/queue/consume	A small HTTP integration suits a team that does not want another language-specific client dependency	The design requires replay, multiple consumer groups, native topic fan-out, a delay beyond seven days, or retention beyond 30 days
RabbitMQ	Consumer acknowledgments provide an explicit point for completion handling	Broker operation becomes part of the system boundary	The requirement is retained event history rather than queued work
BullMQ	It fits a Node.js team that already owns Redis and wants a native producer-worker model	Redis durability and queue operations remain the team's responsibility	Adding Redis only for this weekly path would outweigh the integration benefit
Celery	It fits an existing Python worker estate with an established broker	A Node.js service would add a second runtime and task model	The surrounding worker fleet is not already Python
Sidekiq	It fits a Ruby application that already operates Redis	A Node.js producer-consumer path gains little from a Ruby worker dependency	The application and operations model are not Ruby-based
Kafka	Use it when replay and independent consumer groups are contractual requirements	The team accepts a different operating model to obtain those recovery semantics	The only need is one weekly enqueue-and-process path
Temporal	Use it when digest generation becomes a multi-step workflow with waits or coordination	Workflow state is justified only when the process is more than one job transition	A queue plus a job ledger already expresses the whole process
Airflow	Use it when the work is genuinely a DAG rather than a per-customer message	DAG orchestration adds machinery that a single worker path does not need	Low-latency per-message handling is the primary requirement

The catch for the REST queue option is structural, not cosmetic: there is no DAG or fan-out/join primitive, no native topic one-to-many delivery, and no native debounce or throttle. N queues can simulate multiple recipients, but that changes both operations and cost. Standard delivery remains at-least-once, while FIFO deduplication covers only a five-minute window. A deterministic job claim is therefore still required after the deduplication window.

Stick with Kafka when an operator must replay the last 60 days or several independent consumers must reread the same record. Choose Temporal or Airflow when the digest turns into an orchestrated graph with joins. RabbitMQ remains a credible queue choice when the team explicitly wants its acknowledgment model and is prepared to own the broker boundary. For this property-management batch, a smaller queue-and-ledger design wins only while those exclusions remain true.

Finish the rollout with 3 observable states

First, deploy consumer validation in observation mode against the current envelope and record only safe reason codes, sizes, and schema versions; don't log tenant payloads. Establish the internal byte threshold from the actual serialization path. This is also where the storage architect should verify that referenced private objects remain available through the processing window without being retained longer than the product's deletion obligations permit.

Second, deploy producer rejection and version stamping, then switch new messages to private references. The consumer should accept both explicitly supported versions during the transition and atomically claim the deterministic job ID before generating or sending a digest. Permanently invalid messages go to the DLQ; transient reference-read failures receive bounded retries.

Third, move weekly triggering away from inline generation. A cron execution is limited to 900 seconds, so cron should enqueue work and workers should consume it. Cron targets must be public HTTP URLs, push subscription targets must be public HTTPS URLs, paused schedules do not backfill missed runs, trigger timing has seconds of jitter, and run output retains only its first 4KB. Keep the queryable job ledger as the source of completion state rather than treating cron history as one.

Then stop and measure the right boundary: admission failures by reason code, DLQ dispositions, duplicate job claims, and end-to-end completion against the weekly deadline. Your mileage may vary on the internal size margin, but malformed JSON, unsupported schemas, and payloads beyond 256KB should never reach business logic.

Small messages.

References

• RabbitMQ consumer acknowledgments
• GDPR Article 17