{"record":{"id":"b9e4c6d5d3537af0","repo":"vectordotdev/vector","slug":"pending-messages-limit-of-pending-messages-limit","errorCode":null,"errorMessage":"Pending messages limit of {pending_messages_limit} reached while processing chunk with message id {message_id} and sequence number {sequence_number}","messagePattern":"Pending messages limit of (.+?) reached while processing chunk with message id (.+?) and sequence number (.+?)","errorType":"exception","errorClass":"ChunkedGelfDecoderError","httpStatus":null,"severity":"error","filePath":"lib/codecs/src/decoding/framing/chunked_gelf.rs","lineNumber":378,"sourceCode":"\n        ensure!(\n            sequence_number < total_chunks,\n            InvalidSequenceNumberSnafu {\n                message_id,\n                sequence_number,\n                total_chunks\n            }\n        );\n\n        let mut state_lock = self.state.lock().expect(\"poisoned lock\");\n\n        // Only a new message grows the table, so the limit applies on insert. Checking it\n        // before the lookup rejected chunks of messages already pending, which could then\n        // never complete and expired instead.\n        if !state_lock.contains_key(&message_id)\n            && let Some(pending_messages_limit) = self.pending_messages_limit\n        {\n            ensure!(\n                state_lock.len() < pending_messages_limit,\n                PendingMessagesLimitReachedSnafu {\n                    message_id,\n                    sequence_number,\n                    pending_messages_limit\n                }\n            );\n        }\n\n        let message_state = state_lock.entry(message_id).or_insert_with(|| {\n            // We need to spawn a task that will clear the message state after a certain time\n            // otherwise we will have a memory leak due to messages that never complete\n            let state = Arc::clone(&self.state);\n            let timeout = self.timeout;\n            let timeout_handle = tokio::spawn(async move {\n                tokio::time::sleep(timeout).await;\n                let mut state_lock = state.lock().expect(\"poisoned lock\");\n                if state_lock.remove(&message_id).is_some() {","sourceCodeStart":360,"sourceCodeEnd":396,"githubUrl":"https://github.com/vectordotdev/vector/blob/99894c8d8885659cc16efc39bd71f9fb3b59e296/lib/codecs/src/decoding/framing/chunked_gelf.rs#L360-L396","documentation":"The chunked GELF decoder keeps one reassembly state per in-flight message id and bounds that map with the optional `pending_messages_limit` framing option. When a new chunk arrives while the map already holds `pending_messages_limit` entries, `PendingMessagesLimitReachedSnafu` rejects it — a memory/DoS guard so unfinished messages cannot grow the buffer unboundedly. States expire on their own after `timeout_secs` (default from `ChunkedGelfDecoderConfig`).","triggerScenarios":"Setting `framing.chunked_gelf.pending_messages_limit` (e.g. via GELF/UDP source codec config) to N and then having N message ids pending reassembly — typically heavy chunk loss, a `timeout_secs` longer than the loss recovery window, or a flood of unique message ids (malicious or buggy sender) — while one more chunk arrives.","commonSituations":"Packet loss on high-volume GELF UDP traffic leaving half-assembled messages pinned until timeout; NAT/load-balancer hash changes scattering chunks across Vector replicas so each sees partial messages; raising the limit too low after a traffic increase.","solutions":["Raise `pending_messages_limit` to comfortably above peak concurrent in-flight messages (it is only a safety cap, sized to `rate × timeout_secs`).","Reduce `timeout_secs` (default 5s) so abandoned message states are evicted sooner.","Fix the underlying loss/scattering: increase sender-side chunk reliability, tune NIC/UDP buffers, or ensure affinity so all chunks of a message hit the same Vector instance.","Check whether a hostile or misbehaving sender is spraying unique message ids and rate-limit it at the network edge."],"exampleFix":"# before\ndecoding:\n  codec: gelf\n  framing:\n    method: chunked_gelf\n    chunked_gelf:\n      pending_messages_limit: 1000\n\n# after\ndecoding:\n  codec: gelf\n  framing:\n    method: chunked_gelf\n    chunked_gelf:\n      timeout_secs: 2\n      pending_messages_limit: 50000","handlingStrategy":"retry","validationCode":"# Size the limit before deploy:\n# pending ≈ messages_per_sec × timeout_secs × (loss factor)\n# e.g. 5k msg/s × 5s default timeout × 2 headroom → 50_000\nframing:\n  method: chunked_gelf\n  chunked_gelf:\n    timeout_secs: 5\n    pending_messages_limit: 50000","typeGuard":null,"tryCatchPattern":"Err(e) if e.to_string().contains(\"Pending messages limit\") => {\n    // transient overload: drop the chunk, backpressure the source metrics\n    warn!(error = %e, \"GELF reassembly at capacity; increasing timeout/limit advised\");\n}","preventionTips":["Set pending_messages_limit ≈ 2 × peak rate × timeout_secs rather than a round small number.","Keep timeout_secs as low as your senders' worst retransmit window.","Monitor UDP receive errors and Vector internal metrics to catch loss before the limit trips."],"tags":["gelf","udp","backpressure","memory","limits","vector"],"backgroundTag":"gelf-pending-limit-reached","analyzedSha":"99894c8d8885659cc16efc39bd71f9fb3b59e296","analyzedAt":"2026-08-25T10:53:08.793Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}