Notify track position after skipping malformed packets

playback/src/decoder/mod.rs

@@ -1,3 +1,5 @@
+use std::ops::Deref;
+
 use thiserror::Error;
 
 mod passthrough_decoder;
@@ -54,9 +56,31 @@ impl AudioPacket {
     }
 }
 
+#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
+pub enum AudioPositionKind {
+    // the position is at the expected packet
+    Current,
+    // the decoder skipped some corrupted or invalid data,
+    // and the position is now later than expected
+    SkippedTo,
+}
+
+#[derive(Debug, Clone)]
+pub struct AudioPacketPosition {
+    pub position_ms: u32,
+    pub kind: AudioPositionKind,
+}
+
+impl Deref for AudioPacketPosition {
+    type Target = u32;
+    fn deref(&self) -> &Self::Target {
+        &self.position_ms
+    }
+}
+
 pub trait AudioDecoder {
     fn seek(&mut self, position_ms: u32) -> Result<u32, DecoderError>;
-    fn next_packet(&mut self) -> DecoderResult<Option<(u32, AudioPacket)>>;
+    fn next_packet(&mut self) -> DecoderResult<Option<(AudioPacketPosition, AudioPacket)>>;
 }
 
 impl From<DecoderError> for librespot_core::error::Error {

playback/src/decoder/passthrough_decoder.rs

@@ -7,7 +7,9 @@ use std::{
 // TODO: move this to the Symphonia Ogg demuxer
 use ogg::{OggReadError, Packet, PacketReader, PacketWriteEndInfo, PacketWriter};
 
-use super::{AudioDecoder, AudioPacket, DecoderError, DecoderResult};
+use super::{
+    AudioDecoder, AudioPacket, AudioPacketPosition, AudioPositionKind, DecoderError, DecoderResult,
+};
 
 use crate::{
     metadata::audio::{AudioFileFormat, AudioFiles},
@@ -137,7 +139,7 @@ impl<R: Read + Seek> AudioDecoder for PassthroughDecoder<R> {
         }
     }
 
-    fn next_packet(&mut self) -> DecoderResult<Option<(u32, AudioPacket)>> {
+    fn next_packet(&mut self) -> DecoderResult<Option<(AudioPacketPosition, AudioPacket)>> {
         // write headers if we are (re)starting
         if !self.bos {
             self.wtr
@@ -208,8 +210,14 @@ impl<R: Read + Seek> AudioDecoder for PassthroughDecoder<R> {
 
             if !data.is_empty() {
                 let position_ms = Self::position_pcm_to_ms(pckgp_page);
+                let packet_position = AudioPacketPosition {
+                    position_ms,
+                    kind: AudioPositionKind::Current,
+                };
+
                 let ogg_data = AudioPacket::Raw(std::mem::take(data));
-                return Ok(Some((position_ms, ogg_data)));
+
+                return Ok(Some((packet_position, ogg_data)));
             }
         }
     }

playback/src/decoder/symphonia_decoder.rs

@@ -16,7 +16,9 @@ use symphonia::{
     },
 };
 
-use super::{AudioDecoder, AudioPacket, DecoderError, DecoderResult};
+use super::{
+    AudioDecoder, AudioPacket, AudioPacketPosition, AudioPositionKind, DecoderError, DecoderResult,
+};
 
 use crate::{
     metadata::audio::{AudioFileFormat, AudioFiles},
@@ -170,7 +172,9 @@ impl AudioDecoder for SymphoniaDecoder {
         Ok(self.ts_to_ms(seeked_to_ts.actual_ts))
     }
 
-    fn next_packet(&mut self) -> DecoderResult<Option<(u32, AudioPacket)>> {
+    fn next_packet(&mut self) -> DecoderResult<Option<(AudioPacketPosition, AudioPacket)>> {
+        let mut position_kind = AudioPositionKind::Current;
+
         loop {
             let packet = match self.format.next_packet() {
                 Ok(packet) => packet,
@@ -187,6 +191,10 @@ impl AudioDecoder for SymphoniaDecoder {
             };
 
             let position_ms = self.ts_to_ms(packet.pts());
+            let packet_position = AudioPacketPosition {
+                position_ms,
+                kind: position_kind,
+            };
 
             match self.decoder.decode(&packet) {
                 Ok(decoded) => {
@@ -200,11 +208,14 @@ impl AudioDecoder for SymphoniaDecoder {
                     let sample_buffer = self.sample_buffer.as_mut().unwrap(); // guaranteed above
                     sample_buffer.copy_interleaved_ref(decoded);
                     let samples = AudioPacket::Samples(sample_buffer.samples().to_vec());
-                    return Ok(Some((position_ms, samples)));
+
+                    return Ok(Some((packet_position, samples)));
                 }
                 Err(Error::DecodeError(_)) => {
                     // The packet failed to decode due to corrupted or invalid data, get a new
                     // packet and try again.
+                    warn!("Skipping malformed audio packet at {} ms", position_ms);
+                    position_kind = AudioPositionKind::SkippedTo;
                     continue;
                 }
                 Err(err) => return Err(err.into()),

playback/src/player.rs

@@ -29,7 +29,10 @@ use crate::{
     config::{Bitrate, NormalisationMethod, NormalisationType, PlayerConfig},
     convert::Converter,
     core::{util::SeqGenerator, Error, Session, SpotifyId},
-    decoder::{AudioDecoder, AudioPacket, PassthroughDecoder, SymphoniaDecoder},
+    decoder::{
+        AudioDecoder, AudioPacket, AudioPacketPosition, AudioPositionKind, PassthroughDecoder,
+        SymphoniaDecoder,
+    },
     metadata::audio::{AudioFileFormat, AudioFiles, AudioItem},
     mixer::AudioFilter,
 };
@@ -1103,17 +1106,21 @@ impl Future for PlayerInternal {
                 {
                     match decoder.next_packet() {
                         Ok(result) => {
-                            if let Some((new_stream_position_ms, ref packet)) = result {
+                            if let Some((ref packet_position, ref packet)) = result {
+                                let new_stream_position_ms = packet_position.position_ms;
                                 *stream_position_ms = new_stream_position_ms;
 
                                 if !passthrough {
                                     match packet.samples() {
                                         Ok(_) => {
-                                            let notify_about_position =
+                                            // Only notify if we're skipped some packets *or* we are behind.
-                                                match *reported_nominal_start_time {
+                                            // If we're ahead it's probably due to a buffer of the backend
+                                            // and we're actually in time.
+                                            let notify_about_position = packet_position.kind
+                                                != AudioPositionKind::Current
+                                                || match *reported_nominal_start_time {
                                                     None => true,
                                                     Some(reported_nominal_start_time) => {
-                                                        // only notify if we're behind. If we're ahead it's probably due to a buffer of the backend and we're actually in time.
                                                         let lag = (Instant::now()
                                                             - reported_nominal_start_time)
                                                             .as_millis()
@@ -1340,7 +1347,11 @@ impl PlayerInternal {
         }
     }
 
-    fn handle_packet(&mut self, packet: Option<(u32, AudioPacket)>, normalisation_factor: f64) {
+    fn handle_packet(
+        &mut self,
+        packet: Option<(AudioPacketPosition, AudioPacket)>,
+        normalisation_factor: f64,
+    ) {
         match packet {
             Some((_, mut packet)) => {
                 if !packet.is_empty() {