fix: address code review feedback for AAC encoder PR

- Split wildcard Aac | _ pattern into explicit arms with tracing::warn
  for unrecognised future audio codecs (Critical #1)
- Parameterize DEFAULT_AUDIO_FRAME_DURATION_US by codec: Opus 20ms,
  AAC ~21.333ms via const fn helpers (Suggestion #2)
- Compute AAC timestamps from frame count to avoid truncation drift:
  sequence * 1024 * 1_000_000 / 48_000 (Suggestion #3)
- Document Binary vs EncodedAudio semantic mismatch in AAC encoder
  output pin (Suggestion #4)
- Bundle video/audio codec into MediaCodecConfig struct for
  handle_pin_management (Suggestion #5)
- Deduplicate parse_audio_codec_config: mp4.rs delegates to shared
  implementation in moq/constants.rs (Nit #1)
- Document 960→1024 mixer/encoder frame size interaction and rewrite
  moq_aac_mixing.yml as documented placeholder (Nit #2)

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

Author: streamkit-devin

crates/nodes/src/containers/mp4.rs

@@ -48,8 +48,23 @@ use crate::video::DEFAULT_VIDEO_FRAME_DURATION_US;
 // Constants
 // ---------------------------------------------------------------------------
 
-/// Default audio frame duration when metadata is missing (20 ms Opus frame).
-const DEFAULT_AUDIO_FRAME_DURATION_US: u64 = 20_000;
+/// Default audio frame duration when metadata is missing.
+///
+/// The correct value depends on the codec:
+/// - Opus: 20 ms (960 samples at 48 kHz)
+/// - AAC-LC: ~21.333 ms (1024 samples at 48 kHz)
+///
+/// Use [`default_audio_frame_duration_us`] to get the codec-aware value.
+const DEFAULT_AUDIO_FRAME_DURATION_US_OPUS: u64 = 20_000;
+const DEFAULT_AUDIO_FRAME_DURATION_US_AAC: u64 = 21_333;
+
+/// Return the default audio frame duration for the given codec.
+const fn default_audio_frame_duration_us(codec: AudioCodec) -> u64 {
+    match codec {
+        AudioCodec::Aac => DEFAULT_AUDIO_FRAME_DURATION_US_AAC,
+        _ => DEFAULT_AUDIO_FRAME_DURATION_US_OPUS,
+    }
+}
 
 /// Default video timescale (90 kHz — standard for MPEG transport streams / MP4).
 const DEFAULT_VIDEO_TIMESCALE: NonZeroU32 = match NonZeroU32::new(90_000) {
@@ -608,21 +623,10 @@ fn classify_packet(packet: Packet) -> Option<MuxFrame> {
 
 /// Parse an optional audio codec config string into an [`AudioCodec`].
 ///
-/// Accepted values (case-insensitive): `"opus"`, `"aac"`.
-/// Returns `AudioCodec::Opus` when the input is `None` or unrecognised.
+/// Delegates to the shared [`crate::transport::moq::parse_audio_codec_config`]
+/// parser and defaults to `Opus` when the input is `None` or unrecognised.
 fn parse_mp4_audio_codec_config(s: Option<&str>) -> AudioCodec {
-    match s {
-        Some(v) if v.eq_ignore_ascii_case("aac") => AudioCodec::Aac,
-        Some(v) if v.eq_ignore_ascii_case("opus") => AudioCodec::Opus,
-        Some(other) => {
-            tracing::warn!(
-                audio_codec = other,
-                "unrecognised audio_codec config value, defaulting to Opus"
-            );
-            AudioCodec::Opus
-        },
-        None => AudioCodec::Opus,
-    }
+    s.and_then(crate::transport::moq::parse_audio_codec_config).unwrap_or(AudioCodec::Opus)
 }
 
 /// Determine the MP4 MIME content-type string from optional codec info.
@@ -653,10 +657,19 @@ fn mp4_content_type(audio: Option<AudioCodec>, video: Option<VideoCodec>) -> &'s
         (Some(AudioCodec::Aac), Some(VideoCodec::H264)) => "video/mp4; codecs=\"avc1,mp4a\"",
         // Audio + unknown/future video codec
         (Some(AudioCodec::Opus), Some(_)) => "video/mp4; codecs=\"opus\"",
-        (Some(AudioCodec::Aac | _), Some(_)) => "video/mp4; codecs=\"mp4a\"",
+        (Some(AudioCodec::Aac), Some(_)) => "video/mp4; codecs=\"mp4a\"",
         // Audio-only
         (Some(AudioCodec::Opus), None) => "audio/mp4; codecs=\"opus\"",
-        (Some(AudioCodec::Aac | _), None) => "audio/mp4; codecs=\"mp4a\"",
+        (Some(AudioCodec::Aac), None) => "audio/mp4; codecs=\"mp4a\"",
+        // Future audio codec — warn and omit codecs param.
+        (Some(_), Some(_)) => {
+            tracing::warn!("mp4_content_type: unrecognised audio codec — omitting codecs param");
+            "video/mp4"
+        },
+        (Some(_), None) => {
+            tracing::warn!("mp4_content_type: unrecognised audio codec — omitting codecs param");
+            "audio/mp4"
+        },
         // Video-only
         (None, Some(VideoCodec::Av1)) => "video/mp4; codecs=\"av01\"",
         (None, Some(VideoCodec::H264)) => "video/mp4; codecs=\"avc1\"",
@@ -1213,7 +1226,7 @@ async fn run_stream_mode(
                 let duration_us = metadata
                     .as_ref()
                     .and_then(|m| m.duration_us)
-                    .unwrap_or(DEFAULT_AUDIO_FRAME_DURATION_US);
+                    .unwrap_or_else(|| default_audio_frame_duration_us(session.audio_codec));
                 let duration_ticks = us_to_ticks(duration_us, audio_timescale.get());
 
                 let data_size = data.len();
@@ -1504,7 +1517,13 @@ async fn run_file_mode(
                 )?;
             },
             MuxFrame::Audio(data, metadata) => {
-                process_file_audio_frame(&data, metadata.as_ref(), &mut state, stats_tracker)?;
+                process_file_audio_frame(
+                    &data,
+                    metadata.as_ref(),
+                    session.audio_codec,
+                    &mut state,
+                    stats_tracker,
+                )?;
             },
         }
     }
@@ -1608,14 +1627,16 @@ fn process_file_video_frame(
 fn process_file_audio_frame(
     data: &Bytes,
     metadata: Option<&PacketMetadata>,
+    audio_codec: AudioCodec,
     state: &mut FileMuxState,
     stats_tracker: &mut NodeStatsTracker,
 ) -> Result<(), StreamKitError> {
     state.packet_count += 1;
     stats_tracker.received();
 
-    let duration_us =
-        metadata.and_then(|m| m.duration_us).unwrap_or(DEFAULT_AUDIO_FRAME_DURATION_US);
+    let duration_us = metadata
+        .and_then(|m| m.duration_us)
+        .unwrap_or_else(|| default_audio_frame_duration_us(audio_codec));
     let duration_ticks = us_to_ticks(duration_us, state.audio_timescale.get());
 
     let data_offset = state

crates/nodes/src/transport/moq/constants.rs

@@ -12,7 +12,16 @@ use streamkit_core::types::{
     AudioCodec, EncodedAudioFormat, EncodedVideoFormat, PacketMetadata, PacketType, VideoCodec,
 };
 
-pub const DEFAULT_AUDIO_FRAME_DURATION_US: u64 = 20_000;
+pub const DEFAULT_AUDIO_FRAME_DURATION_US_OPUS: u64 = 20_000;
+pub const DEFAULT_AUDIO_FRAME_DURATION_US_AAC: u64 = 21_333;
+
+/// Return the default audio frame duration for the given codec.
+pub const fn default_audio_frame_duration_us(codec: AudioCodec) -> u64 {
+    match codec {
+        AudioCodec::Aac => DEFAULT_AUDIO_FRAME_DURATION_US_AAC,
+        _ => DEFAULT_AUDIO_FRAME_DURATION_US_OPUS,
+    }
+}
 
 pub fn packet_duration_us(metadata: Option<&PacketMetadata>) -> Option<u64> {
     metadata.and_then(|m| m.duration_us).filter(|d| *d > 0)

crates/nodes/src/transport/moq/mod.rs

@@ -12,6 +12,7 @@
 #![cfg(feature = "moq")]
 
 mod constants;
+pub(crate) use constants::parse_audio_codec_config;
 mod peer;
 mod pull;
 mod push;

crates/nodes/src/transport/moq/peer/config.rs

@@ -89,6 +89,16 @@ pub(super) enum PublisherEvent {
     Disconnected { path: String, error: Option<String> },
 }
 
+/// Resolved codec pair for video and audio.
+///
+/// Bundles the two codec fields that would otherwise be passed as separate
+/// parameters to `handle_pin_management` and friends.
+#[derive(Debug, Clone, Copy)]
+pub(super) struct MediaCodecConfig {
+    pub video: VideoCodec,
+    pub audio: AudioCodec,
+}
+
 /// Media and output configuration shared across subscriber-related functions.
 pub(super) struct SubscriberMediaConfig {
     pub has_video: bool,
@@ -158,6 +168,8 @@ pub(super) struct SubscriberSendCtx<'a> {
     pub last_audio_ts_ms: Option<u64>,
     pub last_video_ts_ms: Option<u64>,
     pub stats_delta_tx: &'a mpsc::Sender<NodeStatsDelta>,
+    /// Resolved audio codec — used for codec-aware default frame durations.
+    pub audio_codec: AudioCodec,
 }
 
 // ── Shared map of dynamic output pin senders ─────────────────────────────────

crates/nodes/src/transport/moq/peer/mod.rs

@@ -18,8 +18,9 @@ pub use config::MoqPeerConfig;
 use config::{
     infer_kind_from_packet, join_gateway_path, make_broadcast_frame, media_kind_for_packet_type,
     normalize_gateway_path, BidirectionalTaskConfig, BroadcastFrame, DynamicOutputs, FrameResult,
-    MediaKind, MediaTypeState, NodeStatsDelta, PublisherEvent, PublisherReceiveLoopWithSlotConfig,
-    SendResult, SubscriberMediaConfig, SubscriberSendCtx, TrackExit,
+    MediaCodecConfig, MediaKind, MediaTypeState, NodeStatsDelta, PublisherEvent,
+    PublisherReceiveLoopWithSlotConfig, SendResult, SubscriberMediaConfig, SubscriberSendCtx,
+    TrackExit,
 };
 
 use crate::transport::moq::constants::{
@@ -639,7 +640,7 @@ impl ProcessorNode for MoqPeerNode {
                         None => std::future::pending().await,
                     }
                 } => {
-                    Self::handle_pin_management(msg, &dynamic_outputs, &subscriber_broadcast_tx, &stats_delta_tx, &shutdown_tx, &mut forwarder_handles, video_codec, audio_codec);
+                    Self::handle_pin_management(msg, &dynamic_outputs, &subscriber_broadcast_tx, &stats_delta_tx, &shutdown_tx, &mut forwarder_handles, MediaCodecConfig { video: video_codec, audio: audio_codec });
                 }
 
                 // Check for shutdown signal
@@ -804,22 +805,20 @@ impl MoqPeerNode {
     ///   respond with an appropriate pin definition.
     /// - [`PinManagementMessage::AddedOutputPin`]: the engine has set up the pin
     ///   distributor and sends us the channel to write frames to.
-    #[allow(clippy::too_many_arguments)]
     fn handle_pin_management(
         msg: PinManagementMessage,
         dynamic_outputs: &DynamicOutputs,
         subscriber_broadcast_tx: &broadcast::Sender<BroadcastFrame>,
         stats_delta_tx: &mpsc::Sender<NodeStatsDelta>,
         shutdown_tx: &broadcast::Sender<()>,
         forwarder_handles: &mut HashMap<String, tokio::task::JoinHandle<()>>,
-        video_codec: VideoCodec,
-        audio_codec: AudioCodec,
+        codecs: MediaCodecConfig,
     ) {
         match msg {
             PinManagementMessage::RequestAddOutputPin { suggested_name, response_tx } => {
                 let pin_name = suggested_name.unwrap_or_else(|| "dynamic_out".to_string());
                 tracing::info!("MoqPeerNode: creating dynamic output pin '{}'", pin_name);
-                let pin = make_dynamic_output_pin(&pin_name, video_codec, audio_codec);
+                let pin = make_dynamic_output_pin(&pin_name, codecs.video, codecs.audio);
                 let _ = response_tx.send(Ok(pin));
             },
             PinManagementMessage::AddedOutputPin { pin, channel } => {
@@ -2128,6 +2127,7 @@ impl MoqPeerNode {
             node_id,
             broadcast_name,
             &stats_delta_tx,
+            media.audio_codec,
         )
         .await?;
 
@@ -2284,6 +2284,7 @@ impl MoqPeerNode {
         node_id: String,
         broadcast_name: String,
         stats_delta_tx: &mpsc::Sender<NodeStatsDelta>,
+        audio_codec: AudioCodec,
     ) -> Result<u64, StreamKitError> {
         let meter = opentelemetry::global::meter("skit_nodes");
         let gap_histogram = meter
@@ -2310,6 +2311,7 @@ impl MoqPeerNode {
             last_audio_ts_ms: None,
             last_video_ts_ms: None,
             stats_delta_tx,
+            audio_codec,
         };
 
         loop {
@@ -2412,7 +2414,9 @@ impl MoqPeerNode {
                 }
 
                 let default_duration = match broadcast_frame.kind {
-                    MediaKind::Audio => super::constants::DEFAULT_AUDIO_FRAME_DURATION_US,
+                    MediaKind::Audio => {
+                        super::constants::default_audio_frame_duration_us(ctx.audio_codec)
+                    },
                     MediaKind::Video => crate::video::DEFAULT_VIDEO_FRAME_DURATION_US,
                 };
                 clock.advance_by_duration_us(broadcast_frame.duration_us, default_duration);
@@ -2556,8 +2560,7 @@ mod tests {
             &stats_delta_tx,
             &shutdown_tx,
             &mut forwarder_handles,
-            VideoCodec::Vp9,
-            AudioCodec::Opus,
+            MediaCodecConfig { video: VideoCodec::Vp9, audio: AudioCodec::Opus },
         );
 
         // If the channel was dropped, try_send would return a closed error.

crates/nodes/src/transport/moq/pull.rs

@@ -4,7 +4,7 @@
 
 //! MoQ Pull Node - subscribes to broadcasts from a MoQ server
 
-use super::constants::DEFAULT_AUDIO_FRAME_DURATION_US;
+use super::constants::DEFAULT_AUDIO_FRAME_DURATION_US_OPUS;
 use crate::video::{AV1_CONTENT_TYPE, H264_CONTENT_TYPE, VP9_CONTENT_TYPE};
 use async_trait::async_trait;
 use bytes::Buf;
@@ -855,7 +855,7 @@ impl MoqPullNode {
                     let (last_ts, default_dur, clock, is_first_in_group) = match source {
                         ReadSource::Audio => (
                             &mut last_audio_timestamp_us,
-                            DEFAULT_AUDIO_FRAME_DURATION_US,
+                            DEFAULT_AUDIO_FRAME_DURATION_US_OPUS,
                             &mut audio_clock,
                             &mut audio_is_first_in_group,
                         ),

crates/nodes/src/transport/moq/push.rs

@@ -5,8 +5,8 @@
 //! MoQ Push Node - publishes packets to a MoQ broadcast
 
 use super::constants::{
-    catalog_audio_codec, catalog_video_codec, moq_accepted_media_types, resolve_audio_codec,
-    resolve_video_codec, DEFAULT_AUDIO_FRAME_DURATION_US,
+    catalog_audio_codec, catalog_video_codec, default_audio_frame_duration_us,
+    moq_accepted_media_types, resolve_audio_codec, resolve_video_codec,
 };
 use async_trait::async_trait;
 use futures::future::poll_fn;
@@ -511,7 +511,7 @@ impl ProcessorNode for MoqPushNode {
                         (
                             &mut audio_clock,
                             &mut audio_seeded,
-                            DEFAULT_AUDIO_FRAME_DURATION_US,
+                            default_audio_frame_duration_us(audio_codec),
                             &mut audio_first_sent,
                             ap,
                         )
@@ -537,7 +537,7 @@ impl ProcessorNode for MoqPushNode {
                         let dur = if state.is_video {
                             crate::video::DEFAULT_VIDEO_FRAME_DURATION_US
                         } else {
-                            DEFAULT_AUDIO_FRAME_DURATION_US
+                            default_audio_frame_duration_us(audio_codec)
                         };
                         (
                             &mut state.clock,

plugins/native/aac-encoder/src/lib.rs

@@ -61,9 +61,7 @@ pub struct AacEncoderNode {
     encoder: shiguredo_fdk_aac::Encoder,
     /// Residual f32 samples that didn't fill a complete 1024×2 frame.
     residual: Vec<f32>,
-    /// Running presentation timestamp in microseconds.
-    timestamp_us: u64,
-    /// Running sequence counter.
+    /// Running sequence counter (also used to compute drift-free timestamps).
     sequence: u64,
     logger: Logger,
 }
@@ -100,19 +98,24 @@ impl AacEncoderNode {
 
     /// Send one encoded AAC frame downstream with timing metadata.
     fn emit_frame(&mut self, data: &[u8], output: &OutputSender) -> Result<(), String> {
+        // Compute timestamp from frame count to avoid accumulating truncation
+        // drift.  1024 samples / 48 000 Hz = 21.333… µs per frame; using
+        // integer arithmetic: sequence * 1024 * 1_000_000 / 48_000.
+        let timestamp_us =
+            (self.sequence as u128 * 1_024 * 1_000_000 / 48_000) as u64;
+
         let packet = Packet::Binary {
             data: bytes::Bytes::copy_from_slice(data),
             content_type: Some(std::borrow::Cow::Borrowed(AAC_CONTENT_TYPE)),
             metadata: Some(PacketMetadata {
-                timestamp_us: Some(self.timestamp_us),
+                timestamp_us: Some(timestamp_us),
                 duration_us: Some(AAC_FRAME_DURATION_US),
                 sequence: Some(self.sequence),
                 keyframe: None,
             }),
         };
         output.send("out", &packet)?;
 
-        self.timestamp_us += AAC_FRAME_DURATION_US;
         self.sequence += 1;
         Ok(())
     }
@@ -140,6 +143,14 @@ impl NativeProcessorNode for AacEncoderNode {
                     }),
                 ],
             )
+            // NOTE: The output type is `PacketType::Binary` rather than
+            // `PacketType::EncodedAudio(Aac)` because the native plugin C ABI
+            // does not yet have a discriminant for `EncodedAudio`.  The
+            // `BinaryWithMeta` transport preserves `content_type` and metadata
+            // so downstream nodes that inspect these fields (e.g. MP4 muxer)
+            // can still identify the codec.  MoQ transport nodes, however,
+            // expect `EncodedAudio` and will reject `Binary` — this is a known
+            // limitation until `EncodedAudio` support is added to the C ABI.
             .output("out", PacketType::Binary)
             .param_schema(serde_json::json!({
                 "type": "object",
@@ -176,7 +187,6 @@ impl NativeProcessorNode for AacEncoderNode {
         Ok(Self {
             encoder,
             residual: Vec::with_capacity(AAC_FRAME_SAMPLES * usize::from(AAC_CHANNELS) * 2),
-            timestamp_us: 0,
             sequence: 0,
             logger,
         })