From 715d44db8e13a599c5b72311f895e6c243617012 Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Thu, 20 Aug 2026 10:03:12 -0400 Subject: [PATCH 1/5] fix(codecs): stop a one-byte GELF message from panicking the source `decode_message` formats the first two bytes of a frame in its trace log, but the guard above only rejects an empty one. A single-byte datagram therefore panics with "range end index 2 out of range", taking the source task down. It needs trace logging enabled for the target, since `tracing` only evaluates the field expression when the level is on, but it is reachable from the same unauthenticated path as everything else the decoder handles. Slice to what is there. The branch already did the right thing otherwise: a short frame is not a framing failure, it is a malformed payload for the deserializer to reject. Introduced in #21816. --- .../src/decoding/framing/chunked_gelf.rs | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index f469e3f5cdb52..a6bfd7e98bf85 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -451,9 +451,11 @@ impl ChunkedGelfDecoder { src.advance(2); self.decode_chunk(src)? } else { + // Slice defensively: a frame here is only known to be non-empty, and one shorter + // than the magic is reachable from an unauthenticated sender. trace!( - "Received an unchunked GELF message. First two bytes of message: {:?}", - &src[0..2] + "Received an unchunked GELF message. First bytes of message: {:?}", + &src[..src.len().min(GELF_MAGIC.len())] ); Some(src) }; @@ -815,6 +817,21 @@ mod tests { )); } + #[rstest] + #[case::one_byte(&b"x"[..])] + #[case::two_bytes(&b"xy"[..])] + #[tokio::test] + #[traced_test] + async fn decode_short_unchunked_frame_does_not_panic(#[case] payload: &[u8]) { + // The trace log on that branch formats two bytes. `traced_test` enables the level. + let mut src = BytesMut::from(payload); + let mut decoder = ChunkedGelfDecoder::default(); + + let frame = decoder.decode_eof(&mut src).expect("must not fail"); + + assert_eq!(frame, Some(Bytes::copy_from_slice(payload))); + } + #[tokio::test] async fn decode_empty_input() { let mut src = BytesMut::new(); From 56a4ddb7c14d301d435ffdfaa78bb1e700b80462 Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Thu, 20 Aug 2026 10:03:56 -0400 Subject: [PATCH 2/5] fix(codecs): apply chunked_gelf pending limit on insert, not to every chunk `pending_messages_limit` was checked before the table lookup, so once the table was full the decoder rejected every chunk, including chunks belonging to messages already pending. Those messages could then never be completed and expired instead, so a burst of new message IDs stalled legitimate in-flight traffic rather than merely capping how much was buffered. Only a new message grows the table, so check it there. --- .../src/decoding/framing/chunked_gelf.rs | 34 ++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index a6bfd7e98bf85..76bfdfb3666f7 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -367,7 +367,12 @@ impl ChunkedGelfDecoder { let mut state_lock = self.state.lock().expect("poisoned lock"); - if let Some(pending_messages_limit) = self.pending_messages_limit { + // Only a new message grows the table, so the limit applies on insert. Checking it + // before the lookup rejected chunks of messages already pending, which could then + // never complete and expired instead. + if !state_lock.contains_key(&message_id) + && let Some(pending_messages_limit) = self.pending_messages_limit + { ensure!( state_lock.len() < pending_messages_limit, PendingMessagesLimitReachedSnafu { @@ -934,6 +939,33 @@ mod tests { assert!(decoder.state.lock().unwrap().len() == 1); } + #[rstest] + #[tokio::test] + async fn decode_accepts_chunks_of_pending_messages_at_the_limit( + two_chunks_message: ([BytesMut; 2], String), + three_chunks_message: ([BytesMut; 3], String), + ) { + // The limit bounds how many messages may be pending, not which chunks are accepted. + let (mut two_chunks, two_chunks_expected) = two_chunks_message; + let (mut three_chunks, _) = three_chunks_message; + let mut decoder = ChunkedGelfDecoder { + pending_messages_limit: Some(1), + ..Default::default() + }; + + let frame = decoder.decode_eof(&mut two_chunks[0]).unwrap(); + assert!(frame.is_none()); + assert_eq!(decoder.state.lock().unwrap().len(), 1); + + // The table is full, so a new message id is rejected. + assert!(decoder.decode_eof(&mut three_chunks[0]).is_err()); + + // ...but the pending message still completes. + let frame = decoder.decode_eof(&mut two_chunks[1]).unwrap(); + assert_eq!(frame, Some(Bytes::from(two_chunks_expected))); + assert_eq!(decoder.state.lock().unwrap().len(), 0); + } + #[rstest] #[tokio::test] async fn decode_chunk_with_different_total_chunks() { From 78eca1205f22a0f56063a59d01c568021166df9d Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Thu, 20 Aug 2026 10:04:35 -0400 Subject: [PATCH 3/5] fix(codecs): abort the timeout task when a chunked_gelf message is dropped Dropping a message for exceeding `max_length` removed its table entry without aborting the timer spawned for it. The task then outlived the entry it was meant to reclaim, so the number of live tasks was not actually bounded by `pending_messages_limit` the way the entry count was. --- .../src/decoding/framing/chunked_gelf.rs | 39 ++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index 76bfdfb3666f7..1e8bb238a207b 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -426,7 +426,11 @@ impl ChunkedGelfDecoder { if let Some(max_length) = self.max_length { let length = message_state.current_length(); if length > max_length { - state_lock.remove(&message_id); + // Abort on removal, or the task outlives its entry and the live-task count is + // no longer bounded by `pending_messages_limit`. + if let Some(dropped) = state_lock.remove(&message_id) { + dropped.timeout_task.abort(); + } return Err(ChunkedGelfDecoderError::MaxLengthExceed { message_id, sequence_number, @@ -966,6 +970,39 @@ mod tests { assert_eq!(decoder.state.lock().unwrap().len(), 0); } + #[rstest] + #[tokio::test(start_paused = true)] + async fn decode_max_length_exceeded_does_not_leak_timeout_task( + two_chunks_message: ([BytesMut; 2], String), + ) { + // An unaborted task outlives its entry, unbounding the live-task count. + let (mut chunks, _) = two_chunks_message; + let mut decoder = ChunkedGelfDecoder { + max_length: Some(5), + ..Default::default() + }; + + assert!(decoder.decode_eof(&mut chunks[0]).unwrap().is_none()); + let timeout_task = { + let state = decoder.state.lock().unwrap(); + state + .values() + .next() + .map(|message_state| message_state.timeout_task.abort_handle()) + .expect("a message should be pending") + }; + assert!(!timeout_task.is_finished()); + + assert!(decoder.decode_eof(&mut chunks[1]).is_err()); + assert_eq!(decoder.state.lock().unwrap().len(), 0); + + tokio::task::yield_now().await; + assert!( + timeout_task.is_finished(), + "the timeout task must be aborted when the message is dropped", + ); + } + #[rstest] #[tokio::test] async fn decode_chunk_with_different_total_chunks() { From be556107433ddc4a17baa17fee3c3ecfc4a4f946 Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Thu, 20 Aug 2026 10:06:17 -0400 Subject: [PATCH 4/5] chore(codecs): stop chunked_gelf pinning the framer's read buffer `BytesDecoder` hands over a slice of `FramedRead`'s buffer without copying, so storing that slice held the whole buffer for as long as the chunk was pending. The buffer is at least 8 KiB, and a chunk sized for an MTU is around 1388 bytes, so a pending message could hold several times what its payload needs. `add_chunk` now copies into an allocation of its own. Two smaller ones alongside it: - Reassembly reserved nothing and grew into a `BytesMut`, so it paid reallocation slack on top of the copy it already needs. Reserve the length up front and release each chunk as it is copied. - `HashMap` stores values inline, and `MessageState` is around 4 KiB because of its fixed chunk array, so every bucket carried one whether occupied or not and a rehash transiently allocated two copies of the table. Box it. --- .../src/decoding/framing/chunked_gelf.rs | 68 +++++++++++++++---- 1 file changed, 54 insertions(+), 14 deletions(-) diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index 1e8bb238a207b..1f3046df064d7 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -137,7 +137,7 @@ impl MessageState { let chunk_bitmap_id = 1 << sequence_number; self.chunks_bitmap |= chunk_bitmap_id; self.current_length += chunk.remaining(); - self.chunks[sequence_number as usize] = chunk; + self.chunks[sequence_number as usize] = Bytes::copy_from_slice(&chunk); } fn is_complete(&self) -> bool { @@ -148,18 +148,20 @@ impl MessageState { self.current_length } - fn retrieve_message(&self) -> Option { - if self.is_complete() { - self.timeout_task.abort(); - let chunks = &self.chunks[0..self.total_chunks as usize]; - let mut message = BytesMut::new(); - for chunk in chunks { - message.extend_from_slice(chunk); - } - Some(message.freeze()) - } else { - None + /// Peak is ~2x the message: a contiguous destination coexists with the chunks it copies + /// from. Reserving exactly keeps a growing buffer from adding slack on top of that. + fn retrieve_message(&mut self) -> Option { + if !self.is_complete() { + return None; } + + self.timeout_task.abort(); + let mut message = BytesMut::with_capacity(self.current_length); + for chunk in &mut self.chunks[0..self.total_chunks as usize] { + message.extend_from_slice(chunk); + *chunk = Bytes::new(); + } + Some(message.freeze()) } } @@ -295,7 +297,7 @@ pub struct ChunkedGelfDecoder { // message, so we have to read all the bytes from the message (datagram) bytes_decoder: BytesDecoder, decompression_config: ChunkedGelfDecompressionConfig, - state: Arc>>, + state: Arc>>>, timeout: Duration, pending_messages_limit: Option, max_length: Option, @@ -399,7 +401,7 @@ impl ChunkedGelfDecoder { ); } }); - MessageState::new(total_chunks, timeout_handle) + Box::new(MessageState::new(total_chunks, timeout_handle)) }); ensure!( @@ -1003,6 +1005,44 @@ mod tests { ); } + #[tokio::test] + async fn add_chunk_does_not_retain_the_source_buffer() { + // `BytesDecoder` slices the read buffer without copying, so retaining it would pin + // the whole buffer (>= 8 KiB) per chunk. + let mut chunk = create_chunk(1u64, 0u8, 2u8, &"foo"); + let source_range = chunk.as_ptr_range(); + let mut decoder = ChunkedGelfDecoder::default(); + + assert!(decoder.decode_eof(&mut chunk).unwrap().is_none()); + + let state = decoder.state.lock().unwrap(); + let message_state = state.values().next().expect("message pending"); + let stored = message_state.chunks[0].as_ptr(); + assert!( + !source_range.contains(&stored), + "the stored chunk must not alias the decoder's input buffer" + ); + } + + #[tokio::test] + async fn retrieve_message_releases_chunks_while_assembling() { + // Does not lower the 2x peak, just shortens how long the source side is held. + let mut state = MessageState::new(2, tokio::spawn(async {})); + state.add_chunk(0, Bytes::from_static(b"foo")); + state.add_chunk(1, Bytes::from_static(b"bar")); + + let message = state + .retrieve_message() + .expect("message should be complete"); + + assert_eq!(message, Bytes::from_static(b"foobar")); + assert!( + state.chunks[..2].iter().all(Bytes::is_empty), + "each chunk must be released as it is copied" + ); + assert_eq!(state.current_length(), 6); + } + #[rstest] #[tokio::test] async fn decode_chunk_with_different_total_chunks() { From 6dc188438c51c58d920ada0d7b34de058ed35504 Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Thu, 20 Aug 2026 10:12:17 -0400 Subject: [PATCH 5/5] chore(changelog): add fragments for the chunked_gelf fixes --- changelog.d/chunked_gelf_limit_handling.fix.md | 3 +++ changelog.d/chunked_gelf_one_byte_panic.fix.md | 3 +++ 2 files changed, 6 insertions(+) create mode 100644 changelog.d/chunked_gelf_limit_handling.fix.md create mode 100644 changelog.d/chunked_gelf_one_byte_panic.fix.md diff --git a/changelog.d/chunked_gelf_limit_handling.fix.md b/changelog.d/chunked_gelf_limit_handling.fix.md new file mode 100644 index 0000000000000..ec92278e162f3 --- /dev/null +++ b/changelog.d/chunked_gelf_limit_handling.fix.md @@ -0,0 +1,3 @@ +Fixed two problems with the `chunked_gelf` framing decoder's limits. `pending_messages_limit` was applied to every chunk rather than only to new messages, so once the limit was reached even chunks of messages already pending were rejected and those messages could never complete. Separately, dropping a message for exceeding `max_length` left its timeout task running, so the number of live tasks was not bounded by `pending_messages_limit` the way the pending message count was. + +authors: pront diff --git a/changelog.d/chunked_gelf_one_byte_panic.fix.md b/changelog.d/chunked_gelf_one_byte_panic.fix.md new file mode 100644 index 0000000000000..0d936fcbe30b9 --- /dev/null +++ b/changelog.d/chunked_gelf_one_byte_panic.fix.md @@ -0,0 +1,3 @@ +Fixed a panic in the `chunked_gelf` framing decoder when a one-byte message arrived and trace-level logging was enabled for it, which took down the source. Such a message is now passed on for the decoder to reject, as any other malformed payload would be. + +authors: pront