rtmp: Type 3 chunks that start a new message don't advance the timestamp

[Saul-Punybz]

Summary

ChunkReader returns the previous header unchanged for every Type 3 chunk. That's correct for a continuation chunk of the same message, but a Type 3 chunk that starts a new message must advance the timestamp by the previous delta (RTMP spec §5.3.1.2.4; FFmpeg's libavformat/rtmppkt.c does the same). As it stands, every message after the second one on a chunk stream keeps the same timestamp when the sender uses Type 3 headers for constant-rate frames.

Affects scuffle-rtmp 0.2.3 (latest on crates.io) and main today (crates/rtmp/src/chunk/reader.rs).

Reproduction

Type 0 (ts 10), Type 1 (delta 40), then two Type 3 chunks that each start a new message:

let mut buf = BytesMut::new();
#[rustfmt::skip]
buf.extend_from_slice(&[
    0b00_000100, 0x00, 0x00, 0x0A, 0x00, 0x00, 0x02, 0x09, 0x01, 0x00, 0x00, 0x00, 0xAA, 0xAA, // type 0, ts 10
    0b01_000100, 0x00, 0x00, 0x28, 0x00, 0x00, 0x02, 0x09, 0xBB, 0xBB,                         // type 1, delta 40
    0b11_000100, 0xCC, 0xCC,                                                                   // type 3, new message
    0b11_000100, 0xDD, 0xDD,                                                                   // type 3, new message
]);
let mut reader = ChunkReader::default();
let ts: Vec<u32> = std::iter::from_fn(|| reader.read_chunk(&mut buf).unwrap())
    .map(|c| c.message_header.timestamp)
    .collect();
assert_eq!(ts, vec![10, 50, 90, 130]); // today: [10, 50, 50, 50]

We hit this in practice with an rml_rtmp client (it sends constant-rate audio/video as Type 3) publishing into a scuffle-rtmp server: every frame after the second had a frozen timestamp.

Proposed fix

Track the last delta per chunk stream (Type 0 → its timestamp, per the spec; Type 1/2 → the delta), committed only when the chunk is actually consumed. For a Type 3 chunk with no partial message pending on that stream, add the delta. Continuation chunks stay as they are. The diff below includes a regression test that also covers a message split across chunks (the continuation must not advance). With it, all existing reader tests still pass.

Diff against crates/rtmp/src/chunk/reader.rs

--- a/crates/rtmp/src/chunk/reader.rs
+++ b/crates/rtmp/src/chunk/reader.rs
@@ -29,6 +29,13 @@
     /// chunk header)
     previous_chunk_headers: HashMap<u32, ChunkMessageHeader>,
 
+    /// the timestamp delta last seen on each chunk stream.
+    /// A Type 3 chunk that starts a new message applies it again (RTMP spec
+    /// 5.3.1.2.4; FFmpeg's `rtmppkt.c` does the same).
+    previous_deltas: HashMap<u32, u32>,
+
     /// Technically according to the spec, we can have multiple message streams
     /// in a single chunk stream. Because of this the key of this map is a tuple
     /// (chunk stream id, message stream id).
@@ -43,6 +50,7 @@
     fn default() -> Self {
         Self {
             previous_chunk_headers: HashMap::with_capacity(MAX_PREVIOUS_CHUNK_HEADERS),
+            previous_deltas: HashMap::with_capacity(MAX_PREVIOUS_CHUNK_HEADERS),
             partial_chunks: HashMap::with_capacity(MAX_PARTIAL_CHUNK_COUNT),
             max_chunk_size: INIT_CHUNK_SIZE,
         }
@@ -166,6 +174,20 @@
                 ));
             }
 
+            // remember the delta a following Type 3 chunk reuses.
+            let delta = match header.format {
+                ChunkType::Type0 => Some(message_header.timestamp),
+                ChunkType::Type1 | ChunkType::Type2 => Some(
+                    self.previous_chunk_headers
+                        .get(&header.chunk_stream_id)
+                        .map_or(message_header.timestamp, |prev| message_header.timestamp.wrapping_sub(prev.timestamp)),
+                ),
+                ChunkType::Type3 => None,
+            };
+            if let Some(delta) = delta {
+                self.previous_deltas.insert(header.chunk_stream_id, delta);
+            }
+
             // We insert the chunk header into our map.
             self.previous_chunk_headers
                 .insert(header.chunk_stream_id, message_header.clone());
@@ -500,6 +522,19 @@
                         .read_u32::<BigEndian>()
                         .eof_to_none()
                         .map_err(|e| e.map(crate::error::RtmpError::Io))?;
+                }
+
+                // a Type 3 chunk that starts a new message (no
+                // partial message pending on this stream) advances the
+                // timestamp by the stored delta; a continuation chunk keeps it.
+                let key = (header.chunk_stream_id, previous_header.msg_stream_id);
+                let continuation = self.partial_chunks.get(&key).is_some_and(|p| !p.is_empty());
+                if !continuation {
+                    let delta = self.previous_deltas.get(&header.chunk_stream_id).copied().unwrap_or(0);
+                    return Ok(ChunkMessageHeader {
+                        timestamp: previous_header.timestamp.wrapping_add(delta),
+                        ..previous_header
+                    });
                 }
 
                 Ok(previous_header)
@@ -1076,4 +1111,44 @@
             assert_eq!(chunk.payload[i], i as u8);
         }
     }
+
+    /// Type 3 chunks that start new messages advance the
+    /// timestamp by the last delta; a Type 3 continuation chunk does not.
+    #[test]
+    fn test_reader_type3_new_messages_reuse_the_delta() {
+        let mut buf = BytesMut::new();
+        #[rustfmt::skip]
+        buf.extend_from_slice(&[
+            0b00_000100, 0x00, 0x00, 0x0A, 0x00, 0x00, 0x02, 0x09, 0x01, 0x00, 0x00, 0x00, 0xAA, 0xAA, // type 0, ts 10
+            0b01_000100, 0x00, 0x00, 0x28, 0x00, 0x00, 0x02, 0x09, 0xBB, 0xBB, // type 1, delta 40
+            0b11_000100, 0xCC, 0xCC, // type 3: new message, delta 40 again
+            0b11_000100, 0xDD, 0xDD, // type 3: new message, delta 40 again
+        ]);
+        let mut reader = ChunkReader::default();
+        let timestamps: Vec<u32> = std::iter::from_fn(|| reader.read_chunk(&mut buf).unwrap())
+            .map(|c| c.message_header.timestamp)
+            .collect();
+        assert_eq!(timestamps, vec![10, 50, 90, 130]);
+
+        // A message split across chunks: the Type 3 continuation keeps the
+        // message's timestamp, and the next new message advances once.
+        let mut buf = BytesMut::new();
+        buf.extend_from_slice(&[0b00_000101, 0x00, 0x00, 0x00, 0x00, 0x00, 0xC8, 0x09, 0x01, 0x00, 0x00, 0x00]);
+        buf.extend_from_slice(&[0u8; 128]);
+        buf.extend_from_slice(&[0b11_000101]);
+        buf.extend_from_slice(&[0u8; 72]);
+        buf.extend_from_slice(&[0b10_000101, 0x00, 0x00, 0x21]);
+        buf.extend_from_slice(&[0u8; 128]);
+        buf.extend_from_slice(&[0b11_000101]);
+        buf.extend_from_slice(&[0u8; 72]);
+        buf.extend_from_slice(&[0b11_000101]);
+        buf.extend_from_slice(&[0u8; 128]);
+        buf.extend_from_slice(&[0b11_000101]);
+        buf.extend_from_slice(&[0u8; 72]);
+        let mut reader = ChunkReader::default();
+        let timestamps: Vec<u32> = std::iter::from_fn(|| reader.read_chunk(&mut buf).unwrap())
+            .map(|c| c.message_header.timestamp)
+            .collect();
+        assert_eq!(timestamps, vec![0, 33, 66]);
+    }
 }

Happy to open a PR if that's easier for you. I didn't here because of the Bazel/Docker setup the contribution guide asks for. The patch is MIT OR Apache-2.0, same as the crate.