For a logistics weekly digest, reject or externalize an oversize message before enqueueing it, validate the same versioned envelope again before doing work, and acknowledge only after the delivery result is durable. That decision rule prevents malformed JSON, schema drift, and a 256KB queue limit from turning one bad customer record into either a retry storm or a silently missed digest.
Short answer: measure the serialized bytes rather than the in-memory object, keep the queue message to identifiers and immutable references, quarantine permanent contract failures, and make the delivery operation idempotent before enabling retries.
I learned the delivery-order part after being paged for missed jobs and duplicate deliveries in production queue systems. The uncomfortable invariant was that a worker can fail on either side of an external effect: acknowledging first can lose the job, while acknowledging afterward can repeat the effect. A valid payload does not remove that ambiguity. It only gets the job far enough to expose it.
This distinction matters for a weekly digest. The schedule may select active customers correctly, yet the queue can still fail later because one producer serialized a large shipment summary, another emitted truncated JSON, or a consumer deployed with a different contract. Don't treat all three as "retry the job." They require different dispositions.
What evidence separates malformed JSON from a 256KB background job queue failure?
A scheduler answers when. The queue contract answers what. The worker answers whether the effect is safe to repeat. Delivery guarantees span all three.
Consider a producer that embeds every shipment event in the message. The object may look reasonable in a debugger, but transport limits apply to encoded bytes. UTF-8 customer names, escaped JSON strings, and a growing shipment array all affect the result. If this queue is configured with a 256KB ceiling, the exact gate is len(serialized) <= 256 * 1024; counting records or characters is only an estimate.
There are four distinct outcomes worth preserving in logs and metrics:
| Failure class | Detected at | Disposition | Retry? |
|---|---|---|---|
| Oversize serialized envelope | Producer admission | Store the digest input elsewhere and enqueue a reference | No, until the envelope changes |
| Malformed JSON | Consumer boundary | Quarantine the original bytes with metadata | No |
| Unsupported schema version or invalid field | Both boundaries | Reject or quarantine with a stable reason code | No |
| Delivery attempt interrupted before durable completion | Worker operation | Redeliver using the same idempotency key | Yes |
The first three are deterministic for the same bytes. Retrying them consumes capacity without changing the answer. The fourth may succeed on another attempt, but it also carries duplicate risk. That's why the idempotency key belongs in the contract rather than being invented by each worker process.
For the digest, a useful key can be derived from stable business identity: customer ID plus the digest period and contract version. The producer should persist that identity with the job. The consumer should record completion against it atomically with whatever durable state represents "this digest was sent." I'm not sure which transaction boundary is available in your mail delivery path; that implementation detail decides whether you can obtain an effectively-once business result or must reconcile ambiguous attempts afterward.
The acknowledgement timeline comes before the fix
Consumer acknowledgements tell the broker when a delivery may be considered handled. RabbitMQ's documentation distinguishes consumer acknowledgements from publisher confirms: they cover different directions and solve different safety problems. A producer confirm does not prove that a customer received a digest, and a consumer acknowledgement does not prove that the original publish was durably accepted.
For this workflow, the worker sequence should be explicit:
- Receive the bytes without acknowledging them.
- Apply the size, parse, and schema gates.
- Quarantine permanent contract failures and then acknowledge that queue delivery.
- Claim the idempotency key or observe an existing durable completion.
- Perform the digest delivery and record its durable outcome.
- Acknowledge only after that outcome is recorded.
Keep it boring.
If the process stops after delivery but before recording completion, redelivery can repeat the external effect. An idempotency key passed through to a delivery provider may close that gap; without support at that boundary, reconciliation is still required. If the process records completion before the effect actually happens, the system can suppress a needed retry. Document the chosen order as an explicit failure matrix, then test process termination between each pair of steps.
Dead-letter or quarantine handling also needs an owner. Alert on rates and oldest age, not every individual bad message. A rising unsupported_schema rate after deployment points toward version skew; a single malformed_json job points toward a specific producer record. Both deserve evidence, but only one should page the on-call engineer.
A byte-first admission path for producer and consumer
Put the same admission function on both sides of the queue. In a Node.js system, the producer still needs to serialize once, measure Buffer.byteLength of those exact bytes, validate the versioned shape, and publish those bytes without re-encoding a different object. The consumer should size-check the received byte buffer before parsing, reject trailing or malformed JSON, and validate the schema version before touching a database or delivery API.
The preventive path below is in Go because the contract is language-neutral and the operational rule is easier to see without framework callbacks. A Node.js producer and consumer should implement these exact states, including the byte-first order.
package digest import ( "bytes" "encoding/json" "errors" "fmt" "io" "time" ) const maxMessageBytes = 256 * 1024 var ( ErrTooLarge = errors.New("payload_too_large") ErrMalformedJSON = errors.New("malformed_json") ErrUnsupportedSchema = errors.New("unsupported_schema") ErrInvalidEnvelope = errors.New("invalid_envelope") ) type DigestJob struct { SchemaVersion int `json:"schema_version"` JobID string `json:"job_id"` CustomerID string `json:"customer_id"` PeriodStart time.Time `json:"period_start"` PeriodEnd time.Time `json:"period_end"` SnapshotRef string `json:"snapshot_ref"` IdempotencyKey string `json:"idempotency_key"` } func EncodeForPublish(job DigestJob) ([]byte, error) { if err := validate(job); err != nil { return nil, err } payload, err := json.Marshal(job) if err != nil { return nil, fmt.Errorf("encode digest job: %w", err) } if len(payload) > maxMessageBytes { return nil, ErrTooLarge } return payload, nil } func DecodeForWork(payload []byte) (DigestJob, error) { if len(payload) > maxMessageBytes { return DigestJob{}, ErrTooLarge } decoder := json.NewDecoder(bytes.NewReader(payload)) decoder.DisallowUnknownFields() var job DigestJob if err := decoder.Decode(&job); err != nil { return DigestJob{}, fmt.Errorf("%w: %v", ErrMalformedJSON, err) } if err := ensureEOF(decoder); err != nil { return DigestJob{}, err } if err := validate(job); err != nil { return DigestJob{}, err } return job, nil } func ensureEOF(decoder *json.Decoder) error { var extra any if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) { return ErrMalformedJSON } return nil } func validate(job DigestJob) error { if job.SchemaVersion != 1 { return ErrUnsupportedSchema } if job.JobID == "" || job.CustomerID == "" || job.SnapshotRef == "" || job.IdempotencyKey == "" || !job.PeriodStart.Before(job.PeriodEnd) { return ErrInvalidEnvelope } return nil } One nuance is deliberate: unknown fields fail admission. That is appropriate when producers and consumers deploy under a controlled contract and an accidental rename must stop. It is not universally correct. If independent producers need forward compatibility, define an explicit compatibility policy and test it; silently accepting everything makes schema evolution look safe until a newly required semantic is absent.
Keep reason codes low-cardinality (payload_too_large, malformed_json, unsupported_schema, invalid_envelope) and keep customer IDs out of metric labels. Put job ID, schema version, payload byte count, producer build, and quarantine reference in structured logs with access controls. The raw digest can contain personal data, so it should not be copied casually into logs.
Compare inline messages with immutable snapshot references
An envelope with snapshot_ref changes the failure surface. The weekly scheduler freezes the customer and shipment selection for a stated period, writes that snapshot durably, and publishes a small job containing identity, bounds, version, and reference. A retry reads the same immutable snapshot instead of rerunning a query against moving data.
This is the long paragraph I want in the runbook because it catches a common half-fix. Moving the large array out of the queue is not enough if snapshot_ref points to a mutable object, expires before the queue's maximum retry window, or can be replaced under the same name. The producer must finish the object write before publishing the reference; the consumer must verify that the reference resolves to the expected tenant and period; retention must cover scheduling delay, retries, incident investigation, and replay; and deletion must honor the applicable data policy. GDPR Article 17 establishes a right to erasure in specified circumstances, so a team handling covered personal data needs a deletion path for snapshots, quarantine records, idempotency records, and backups rather than assuming that an opaque object key removes the obligation. Legal scope and retention periods belong with the responsible privacy and legal teams, not in a worker constant.
The catch is extra lifecycle machinery. A reference message is not suitable when the payload is already small, non-sensitive, and truly self-contained, or when workers cannot reliably access shared storage. In that case, keep the data inline but retain byte admission, schema versioning, and idempotency. Conversely, stick with a reference when input growth is unbounded or replay must use the exact historical selection.
No magic here.
Rehearse the boundary conditions before deployment
Unit tests should feed the decoder an empty body, truncated JSON, a second JSON value after a valid object, unknown fields, unsupported versions, missing identifiers, inverted period bounds, exactly 262,144 bytes, and 262,145 bytes. The boundary deserves literal test fixtures. Otherwise, someone will accidentally compare characters, use < instead of <=, or measure before the final serialization.
Then run a compatibility test between the currently deployed consumer and the next producer build, plus the currently deployed producer and the next consumer build. Roll out consumers that understand a new version before producers emit it. During deployment, graph accepted jobs, quarantines by reason, redeliveries, idempotency conflicts, oldest ready-job age, and end-to-end digest completion by period. A green publish counter alone misses the customer outcome.
This advice is excessive for an ephemeral best-effort cache refresh where duplicates and drops have no material consequence. It is also insufficient for a workflow whose external side effect cannot be deduplicated or reconciled. For the logistics digest, the correct choice depends on the promise made to customers: if one digest per active customer per week matters, retain the immutable input and completion ledger long enough to prove or repair that promise.
