fix(nodes): derive codec metadata from actual encoder/decoder output

- VA-API H.264 encoder: parse NAL unit types to detect IDR keyframes
  instead of cloning input metadata (which blindly propagated
  video::colorbars keyframe=true to every packet). Also derive
  timestamp from CodedBitstreamBuffer metadata. flush_encoder() now
  emits proper metadata instead of None.

- VA-API AV1 encoder: scan OBU headers for Sequence Header to detect
  keyframes instead of cloning input metadata. Same timestamp and
  flush_encoder() fixes as H.264.

- VA-API H.264 decoder: use DecodedHandle::timestamp() for output
  frame metadata instead of cloning the current input packet's
  metadata. Fixes incorrect timestamps on B-frame reordered output.

- Vulkan Video decoder: always replace timestamp_us with the decoder
  output PTS (not just when input metadata had no timestamp). Create
  fresh metadata from decoder PTS when input had none.

- NV-AV1 encoder: remove frame-ordinal next_pts that was incorrectly
  used as timestamp_us when input had no metadata. Downstream
  consumers expect real microsecond timestamps, not counters.

- VA-API NV12 helpers: add bounds checks on UV-plane slice access to
  prevent panics on short buffers. Add up-front buffer size validation
  in write_nv12_to_mapping. Add debug_assert on plane count in
  dma_frame_layout. Fix codec-specific error message in shared helper.

Fixes keyframe flag propagation for downstream transport::rtmp,
containers::mp4, and transport::moq::peer consumers. Fixes timestamp
ordering for H.264 streams with B-frame reordering.

Signed-off-by: StreamKit Devin <devin@streamkit.dev>
Co-Authored-By: Claudio Costa <cstcld91@gmail.com>

Author: streamkit-devin

crates/nodes/src/video/nv_av1.rs

@@ -513,7 +513,6 @@ impl EncoderNodeRunner for NvAv1EncoderNode {
 
 struct NvAv1Encoder {
     encoder: shiguredo_nvcodec::Encoder,
-    next_pts: i64,
 }
 
 impl StandardVideoEncoder for NvAv1Encoder {
@@ -558,7 +557,7 @@ impl StandardVideoEncoder for NvAv1Encoder {
             "NVENC AV1 encoder created"
         );
 
-        Ok(Self { encoder, next_pts: 0 })
+        Ok(Self { encoder })
     }
 
     fn encode(
@@ -609,11 +608,8 @@ impl NvAv1Encoder {
             );
             let data = Bytes::from(encoded.into_data());
 
-            let pts = self.next_pts;
-            self.next_pts += 1;
-
             let meta = remaining_metadata.take();
-            let output_metadata = merge_keyframe_metadata(meta, is_keyframe, pts);
+            let output_metadata = merge_keyframe_metadata(meta, is_keyframe);
 
             packets.push(EncodedPacket { data, metadata: Some(output_metadata) });
         }
@@ -664,16 +660,16 @@ fn i420_to_nv12_buffer(frame: &VideoFrame) -> Vec<u8> {
     nv12
 }
 
+/// Build output metadata from the input metadata (if any) with the correct
+/// keyframe flag from the encoder output.  When input metadata is `None`
+/// (e.g. during `flush_encoder`), the timestamp is left as `None` rather
+/// than using a frame ordinal — downstream consumers expect real
+/// microsecond timestamps, not counters.
 #[allow(clippy::missing_const_for_fn)] // map_or with closures is not yet stable in const fn
-fn merge_keyframe_metadata(
-    metadata: Option<PacketMetadata>,
-    keyframe: bool,
-    pts: i64,
-) -> PacketMetadata {
+fn merge_keyframe_metadata(metadata: Option<PacketMetadata>, keyframe: bool) -> PacketMetadata {
     metadata.map_or(
         PacketMetadata {
-            #[allow(clippy::cast_sign_loss)]
-            timestamp_us: if pts >= 0 { Some(pts as u64) } else { None },
+            timestamp_us: None,
             duration_us: None,
             sequence: None,
             keyframe: Some(keyframe),

crates/nodes/src/video/vaapi_av1.rs

@@ -313,7 +313,14 @@ pub(super) fn upload_nv12_to_dma_frame(
             }
             let uv_h = h.div_ceil(2);
             let chroma_row_bytes = w.div_ceil(2) * 2;
-            let src_uv = &src[w * h..];
+            let y_size = w * h;
+            if src.len() < y_size {
+                return Err(format!(
+                    "NV12 source buffer too short for UV plane: {len} < {y_size} (Y size)",
+                    len = src.len(),
+                ));
+            }
+            let src_uv = &src[y_size..];
             for row in 0..uv_h {
                 let s = row * chroma_row_bytes;
                 let d = uv_offset_img + row * uv_pitch;
@@ -413,6 +420,16 @@ fn dma_frame_layout(
     coded_width: u32,
     coded_height: u32,
 ) -> FrameLayout {
+    // This layout computation assumes a single contiguous buffer object
+    // (all planes share buffer_index 0).  Multi-BO configurations would
+    // need per-plane buffer_index values derived from the DMA-BUF handles.
+    // NV12 has exactly 2 planes (Y + interleaved UV).
+    debug_assert!(
+        CrosVideoFrame::num_planes(dma_frame) == 2,
+        "dma_frame_layout expects 2-plane NV12 but frame has {} planes",
+        CrosVideoFrame::num_planes(dma_frame),
+    );
+
     let pitches = CrosVideoFrame::get_plane_pitch(dma_frame);
     let sizes = CrosVideoFrame::get_plane_size(dma_frame);
 
@@ -511,18 +528,21 @@ pub(super) fn read_nv12_from_mapping(
     let mut data = vec![0u8; y_size + uv_size];
 
     // Y plane.
-    if !planes.is_empty() {
-        let y_stride = plane_pitches.first().copied().unwrap_or(w);
-        if y_stride == w {
-            let copy_len = y_size.min(planes[0].len());
-            data[..copy_len].copy_from_slice(&planes[0][..copy_len]);
-        } else {
-            for row in 0..h {
-                let dst_off = row * w;
-                let src_off = row * y_stride;
-                if src_off + w <= planes[0].len() && dst_off + w <= y_size {
-                    data[dst_off..dst_off + w].copy_from_slice(&planes[0][src_off..src_off + w]);
-                }
+    if planes.is_empty() {
+        tracing::warn!(width, height, "read_nv12_from_mapping: mapping returned no planes");
+        return data;
+    }
+
+    let y_stride = plane_pitches.first().copied().unwrap_or(w);
+    if y_stride == w {
+        let copy_len = y_size.min(planes[0].len());
+        data[..copy_len].copy_from_slice(&planes[0][..copy_len]);
+    } else {
+        for row in 0..h {
+            let dst_off = row * w;
+            let src_off = row * y_stride;
+            if src_off + w <= planes[0].len() && dst_off + w <= y_size {
+                data[dst_off..dst_off + w].copy_from_slice(&planes[0][src_off..src_off + w]);
             }
         }
     }
@@ -567,6 +587,21 @@ pub(super) fn write_nv12_to_mapping(
     let h = frame.height as usize;
     let src = frame.data.as_ref().as_ref();
 
+    // Validate minimum source buffer size up-front so we return an error
+    // rather than silently producing a partial (black-row) frame.
+    let expected_min = match frame.pixel_format {
+        PixelFormat::Nv12 => w * h + w.div_ceil(2) * 2 * h.div_ceil(2),
+        PixelFormat::I420 => w * h + 2 * (w.div_ceil(2) * h.div_ceil(2)),
+        _ => 0, // handled by the match-arm error below
+    };
+    if expected_min > 0 && src.len() < expected_min {
+        return Err(format!(
+            "source frame buffer too short for {:?} {w}×{h}: {len} < {expected_min}",
+            frame.pixel_format,
+            len = src.len(),
+        ));
+    }
+
     match frame.pixel_format {
         PixelFormat::Nv12 => {
             let y_size = w * h;
@@ -597,6 +632,12 @@ pub(super) fn write_nv12_to_mapping(
             if planes.len() > 1 {
                 let uv_stride = plane_pitches.get(1).copied().unwrap_or(chroma_w);
                 let mut uv_plane = planes[1].borrow_mut();
+                if src.len() < y_size {
+                    return Err(format!(
+                        "NV12 source buffer too short for UV plane: {len} < {y_size} (Y size)",
+                        len = src.len(),
+                    ));
+                }
                 let src_uv = &src[y_size..];
                 if uv_stride == chroma_w {
                     let n = uv_size.min(uv_plane.len()).min(src_uv.len());
@@ -656,7 +697,7 @@ pub(super) fn write_nv12_to_mapping(
         },
         _ => {
             return Err(format!(
-                "VA-API AV1 encoder requires NV12 or I420 input, got {:?}",
+                "VA-API encoder requires NV12 or I420 input, got {:?}",
                 frame.pixel_format
             ));
         },
@@ -1284,9 +1325,12 @@ impl StandardVideoEncoder for VaapiAv1Encoder {
         loop {
             match self.encoder.poll() {
                 Ok(Some(coded)) => {
+                    let is_key = av1_bitstream_is_keyframe(&coded.bitstream);
+                    let out_meta =
+                        build_av1_encoder_output_metadata(metadata.clone(), is_key, &coded);
                     packets.push(EncodedPacket {
                         data: Bytes::from(coded.bitstream),
-                        metadata: metadata.clone(),
+                        metadata: Some(out_meta),
                     });
                 },
                 Ok(None) => break,
@@ -1304,8 +1348,12 @@ impl StandardVideoEncoder for VaapiAv1Encoder {
         loop {
             match self.encoder.poll() {
                 Ok(Some(coded)) => {
-                    packets
-                        .push(EncodedPacket { data: Bytes::from(coded.bitstream), metadata: None });
+                    let is_key = av1_bitstream_is_keyframe(&coded.bitstream);
+                    let out_meta = build_av1_encoder_output_metadata(None, is_key, &coded);
+                    packets.push(EncodedPacket {
+                        data: Bytes::from(coded.bitstream),
+                        metadata: Some(out_meta),
+                    });
                 },
                 Ok(None) => break,
                 Err(e) => return Err(format!("VA-API AV1 encoder poll error: {e}")),
@@ -1320,6 +1368,88 @@ impl StandardVideoEncoder for VaapiAv1Encoder {
     }
 }
 
+// ---------------------------------------------------------------------------
+// Encoder metadata helpers
+// ---------------------------------------------------------------------------
+
+use cros_codecs::encoder::CodedBitstreamBuffer;
+
+/// Returns `true` if the AV1 bitstream contains a Sequence Header OBU
+/// (type 1), which the cros-codecs AV1 encoder emits only at keyframes.
+///
+/// The `CodedBitstreamBuffer::metadata.force_keyframe` only reflects the
+/// *input* request — not the actual coded frame type.  Scanning for a
+/// Sequence Header OBU gives ground truth because the cros-codecs AV1
+/// predictor emits one exclusively for keyframes (both forced and periodic).
+fn av1_bitstream_is_keyframe(bitstream: &[u8]) -> bool {
+    const OBU_SEQUENCE_HEADER: u8 = 1;
+
+    let mut pos = 0;
+    while pos < bitstream.len() {
+        let header = bitstream[pos];
+        let obu_type = (header >> 3) & 0x0F;
+        let has_extension = (header >> 2) & 1 != 0;
+        let has_size = (header >> 1) & 1 != 0;
+        pos += 1;
+
+        if obu_type == OBU_SEQUENCE_HEADER {
+            return true;
+        }
+
+        if has_extension {
+            if pos >= bitstream.len() {
+                break;
+            }
+            pos += 1;
+        }
+
+        if has_size {
+            // Parse LEB128 size field.
+            let mut size: usize = 0;
+            for shift in 0..8 {
+                if pos >= bitstream.len() {
+                    return false;
+                }
+                let byte = bitstream[pos];
+                pos += 1;
+                size |= usize::from(byte & 0x7F) << (shift * 7);
+                if byte & 0x80 == 0 {
+                    break;
+                }
+            }
+            pos += size;
+        } else {
+            // Without a size field the OBU extends to the end.
+            break;
+        }
+    }
+    false
+}
+
+/// Build output `PacketMetadata` from the coded bitstream rather than
+/// blindly cloning the input metadata.  The keyframe flag is determined
+/// by OBU-header inspection and the timestamp comes from the coded
+/// buffer's `FrameMetadata`.
+fn build_av1_encoder_output_metadata(
+    input_meta: Option<PacketMetadata>,
+    is_keyframe: bool,
+    coded: &CodedBitstreamBuffer,
+) -> PacketMetadata {
+    match input_meta {
+        Some(mut m) => {
+            m.keyframe = Some(is_keyframe);
+            m.timestamp_us = Some(coded.metadata.timestamp);
+            m
+        },
+        None => PacketMetadata {
+            timestamp_us: Some(coded.metadata.timestamp),
+            duration_us: None,
+            sequence: None,
+            keyframe: Some(is_keyframe),
+        },
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Registration
 // ---------------------------------------------------------------------------