fix(session): resume Qobuz streams at saved position without losing seek (#315)
Three bugs were stacking up to make session resume drop the saved offset
for streamed Qobuz tracks:
1. Position save was debounced, not throttled. Each playback tick reset
the 5s timer, so the save never fired while a track played
continuously — the persisted position froze at whatever value got
written during the first quiet window after track start (typically
~2s). Switch to throttle: fire the first call immediately, then once
per 5s window, with a tail call to flush the latest value when the
window ends.
2. togglePlay tried to apply the restore offset via a separate cmdSeek
right after cmdPlayTrack. For Qobuz streaming the audio thread's
Seek handler bails when current_audio_data is unset — streaming
uses current_streaming_source instead — so the seek was silently
dropped and the track started at 0:00. Pass start_position_secs
directly to v2_play_track. The streaming backend (legacy + V2
player) waits for enough buffer to cover the offset plus 8s
headroom, then eagerly decode-and-discards decoder output up to
the offset before engaging the audio engine — no race, no
first-pull underrun. Plex and local tracks keep the old play+seek
path; they load full audio_data so Seek works as-is.
3. The seekbar showed 0:00 from app launch until the user hit play,
even when a saved position was queued. setPendingSessionRestore
now pre-positions currentTime when positionSecs > 0 so the UI
reflects "paused at X" the moment the app opens.
Other plumbing:
- AudioCommand::PlayStreaming carries start_position_secs and
content_length (used to estimate bytes-per-second for the resume
buffer target).
- Player::play_streaming{,_dynamic} grow a start_position_secs
parameter; QConnect and api_server callers pass 0 (remote-driven
playback always starts fresh).
- cmdPlayTrack forwards startPositionSecs as a 5th optional arg.
cc
This commit is contained in:
8 files changed
+331
-66
No files matched your search
@@ -68,6 +68,13 @@ enum AudioCommand {
|
|||||||
sample_rate: u32,
|
sample_rate: u32,
|
||||||
channels: u16,
|
channels: u16,
|
||||||
duration_secs: u64,
|
duration_secs: u64,
|
||||||
|
/// Resume offset in seconds (#315). When > 0, the audio thread
|
||||||
|
/// waits for enough buffer to cover the offset and pre-skips
|
||||||
|
/// decoder output up to that point before engaging audio.
|
||||||
|
start_position_secs: u64,
|
||||||
|
/// Total content length in bytes. Combined with `duration_secs`
|
||||||
|
/// to estimate bytes-per-second when sizing the resume buffer.
|
||||||
|
content_length: u64,
|
||||||
},
|
},
|
||||||
/// Pause playback
|
/// Pause playback
|
||||||
Pause,
|
Pause,
|
||||||
@@ -1943,13 +1950,16 @@ impl Player {
|
|||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
start_position_secs,
|
||||||
|
content_length,
|
||||||
} => {
|
} => {
|
||||||
log::info!(
|
log::info!(
|
||||||
"Audio thread: starting streaming playback for track {} ({}Hz, {} channels, {}s)",
|
"Audio thread: starting streaming playback for track {} ({}Hz, {} channels, {}s, start={}s)",
|
||||||
track_id,
|
track_id,
|
||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs
|
duration_secs,
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
*pause_suspend_deadline = None;
|
*pause_suspend_deadline = None;
|
||||||
|
|
||||||
@@ -2171,12 +2181,44 @@ impl Player {
|
|||||||
let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0;
|
let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0;
|
||||||
apply_engine_volume(&stream_opt, &engine, volume);
|
apply_engine_volume(&stream_opt, &engine, volume);
|
||||||
|
|
||||||
// Wait for minimum buffer before starting playback
|
// Wait for minimum buffer before starting playback.
|
||||||
|
// When start_position_secs > 0 (session resume),
|
||||||
|
// also wait for enough buffer to cover the resume
|
||||||
|
// offset plus an 8s headroom — the eager pre-skip
|
||||||
|
// below decodes-and-discards up to the offset and
|
||||||
|
// needs the bytes available without blocking the
|
||||||
|
// audio device on the first pull.
|
||||||
log::info!("Streaming: waiting for initial buffer...");
|
log::info!("Streaming: waiting for initial buffer...");
|
||||||
let start_wait = Instant::now();
|
let start_wait = Instant::now();
|
||||||
let max_wait = Duration::from_secs(30);
|
let max_wait = Duration::from_secs(60);
|
||||||
|
|
||||||
while !source.has_min_buffer() && start_wait.elapsed() < max_wait {
|
let bytes_per_sec_estimate: u64 = if duration_secs > 0
|
||||||
|
&& content_length > 0
|
||||||
|
{
|
||||||
|
content_length / duration_secs
|
||||||
|
} else {
|
||||||
|
200_000
|
||||||
|
};
|
||||||
|
let resume_buffer_target: u64 = if start_position_secs > 0 {
|
||||||
|
bytes_per_sec_estimate
|
||||||
|
.saturating_mul(start_position_secs.saturating_add(8))
|
||||||
|
} else {
|
||||||
|
0
|
||||||
|
};
|
||||||
|
|
||||||
|
let buffer_sufficient = |src: &Arc<BufferedMediaSource>| -> bool {
|
||||||
|
if !src.has_min_buffer() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if resume_buffer_target == 0 {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
(src.buffer_size() as u64) >= resume_buffer_target
|
||||||
|
};
|
||||||
|
|
||||||
|
while !buffer_sufficient(&source)
|
||||||
|
&& start_wait.elapsed() < max_wait
|
||||||
|
{
|
||||||
std::thread::sleep(Duration::from_millis(50));
|
std::thread::sleep(Duration::from_millis(50));
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2184,11 +2226,20 @@ impl Player {
|
|||||||
log::error!("Streaming: timeout waiting for initial buffer");
|
log::error!("Streaming: timeout waiting for initial buffer");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if resume_buffer_target > 0
|
||||||
|
&& (source.buffer_size() as u64) < resume_buffer_target
|
||||||
|
{
|
||||||
|
log::warn!(
|
||||||
|
"Streaming: timed out waiting for resume buffer (got {} bytes, wanted {}); pre-skip may underrun briefly",
|
||||||
|
source.buffer_size(),
|
||||||
|
resume_buffer_target
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
let buffer_wait_ms = start_wait.elapsed().as_millis();
|
let buffer_wait_ms = start_wait.elapsed().as_millis();
|
||||||
log::info!(
|
log::info!(
|
||||||
"Streaming: initial buffer ready in {}ms, creating incremental decoder...",
|
"Streaming: buffer ready in {}ms ({} bytes, target {}), creating incremental decoder...",
|
||||||
buffer_wait_ms
|
buffer_wait_ms, source.buffer_size(), resume_buffer_target
|
||||||
);
|
);
|
||||||
|
|
||||||
// Create incremental streaming source - this starts playback IMMEDIATELY
|
// Create incremental streaming source - this starts playback IMMEDIATELY
|
||||||
@@ -2266,8 +2317,40 @@ impl Player {
|
|||||||
thread_state.set_normalization_gain(normalization);
|
thread_state.set_normalization_gain(normalization);
|
||||||
|
|
||||||
// Box the incremental source to match the expected type
|
// Box the incremental source to match the expected type
|
||||||
let source_to_play: Box<dyn Source<Item = f32> + Send> =
|
let mut source_to_play: Box<dyn Source<Item = f32> + Send> =
|
||||||
Box::new(incremental_source);
|
Box::new(incremental_source);
|
||||||
|
|
||||||
|
// Eager pre-skip for session resume. Decode and
|
||||||
|
// discard samples here so the engine's first pull
|
||||||
|
// doesn't have to do the work synchronously, which
|
||||||
|
// would underrun the audio device for multi-second
|
||||||
|
// offsets. The buffer wait above guarantees
|
||||||
|
// there's enough downloaded data to feed this loop.
|
||||||
|
if start_position_secs > 0 {
|
||||||
|
let target_samples: u64 = (start_position_secs)
|
||||||
|
.saturating_mul(actual_sr as u64)
|
||||||
|
.saturating_mul(actual_ch as u64);
|
||||||
|
let skip_start = Instant::now();
|
||||||
|
let mut skipped: u64 = 0;
|
||||||
|
while skipped < target_samples {
|
||||||
|
if source_to_play.next().is_none() {
|
||||||
|
log::warn!(
|
||||||
|
"Resume: source ended before reaching {}s (pre-skipped {} samples)",
|
||||||
|
start_position_secs,
|
||||||
|
skipped
|
||||||
|
);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
skipped += 1;
|
||||||
|
}
|
||||||
|
log::info!(
|
||||||
|
"Resume: pre-skipped {} samples ({}s) in {}ms",
|
||||||
|
skipped,
|
||||||
|
start_position_secs,
|
||||||
|
skip_start.elapsed().as_millis()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
// Wrap source with diagnostic, normalization, and visualizer
|
// Wrap source with diagnostic, normalization, and visualizer
|
||||||
let source_to_play = wrap_source(
|
let source_to_play = wrap_source(
|
||||||
source_to_play,
|
source_to_play,
|
||||||
@@ -2282,16 +2365,19 @@ impl Player {
|
|||||||
}
|
}
|
||||||
|
|
||||||
thread_state.is_playing.store(true, Ordering::SeqCst);
|
thread_state.is_playing.store(true, Ordering::SeqCst);
|
||||||
thread_state.position.store(0, Ordering::SeqCst);
|
thread_state
|
||||||
|
.position
|
||||||
|
.store(start_position_secs, Ordering::SeqCst);
|
||||||
thread_state
|
thread_state
|
||||||
.current_track_id
|
.current_track_id
|
||||||
.store(track_id, Ordering::SeqCst);
|
.store(track_id, Ordering::SeqCst);
|
||||||
thread_state.start_playback_timer(0);
|
thread_state.start_playback_timer(start_position_secs);
|
||||||
|
|
||||||
*current_engine = Some(engine);
|
*current_engine = Some(engine);
|
||||||
log::info!(
|
log::info!(
|
||||||
"Audio thread: streaming playback STARTED in {}ms (incremental decode active)",
|
"Audio thread: streaming playback STARTED in {}ms at {}s (incremental decode active)",
|
||||||
start_wait.elapsed().as_millis()
|
start_wait.elapsed().as_millis(),
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
AudioCommand::Pause => {
|
AudioCommand::Pause => {
|
||||||
@@ -3301,8 +3387,11 @@ impl Player {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Play from streaming source (starts playback before full download)
|
/// Play from streaming source (starts playback before full download).
|
||||||
/// Returns the BufferWriter so caller can push data as it downloads
|
/// Returns the BufferWriter so caller can push data as it downloads.
|
||||||
|
/// `start_position_secs` > 0 turns this into a session-resume play
|
||||||
|
/// (#315): the audio thread waits for enough buffer to cover the
|
||||||
|
/// offset and pre-skips decoder output up to that point.
|
||||||
pub fn play_streaming(
|
pub fn play_streaming(
|
||||||
&self,
|
&self,
|
||||||
track_id: u64,
|
track_id: u64,
|
||||||
@@ -3311,14 +3400,16 @@ impl Player {
|
|||||||
content_length: u64,
|
content_length: u64,
|
||||||
buffer_seconds: u8,
|
buffer_seconds: u8,
|
||||||
duration_secs: u64,
|
duration_secs: u64,
|
||||||
|
start_position_secs: u64,
|
||||||
) -> Result<BufferWriter, String> {
|
) -> Result<BufferWriter, String> {
|
||||||
log::info!(
|
log::info!(
|
||||||
"Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total, {}s)",
|
"Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total, {}s, start={}s)",
|
||||||
track_id,
|
track_id,
|
||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
content_length,
|
content_length,
|
||||||
duration_secs
|
duration_secs,
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
|
|
||||||
// Use StreamingConfig::from_seconds for proper buffer sizing
|
// Use StreamingConfig::from_seconds for proper buffer sizing
|
||||||
@@ -3334,6 +3425,8 @@ impl Player {
|
|||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
start_position_secs,
|
||||||
|
content_length,
|
||||||
})
|
})
|
||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
log::error!("Player: Failed to send streaming command: {}", e);
|
log::error!("Player: Failed to send streaming command: {}", e);
|
||||||
@@ -3344,8 +3437,8 @@ impl Player {
|
|||||||
Ok(writer)
|
Ok(writer)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Play from streaming source with dynamic buffer based on measured speed
|
/// Play from streaming source with dynamic buffer based on measured speed.
|
||||||
/// Returns the BufferWriter so caller can push data as it downloads
|
/// `start_position_secs` > 0 signals session resume (see `play_streaming`).
|
||||||
pub fn play_streaming_dynamic(
|
pub fn play_streaming_dynamic(
|
||||||
&self,
|
&self,
|
||||||
track_id: u64,
|
track_id: u64,
|
||||||
@@ -3355,16 +3448,18 @@ impl Player {
|
|||||||
content_length: u64,
|
content_length: u64,
|
||||||
speed_mbps: f64,
|
speed_mbps: f64,
|
||||||
duration_secs: u64,
|
duration_secs: u64,
|
||||||
|
start_position_secs: u64,
|
||||||
) -> Result<BufferWriter, String> {
|
) -> Result<BufferWriter, String> {
|
||||||
log::info!(
|
log::info!(
|
||||||
"Player: Starting dynamic streaming for track {} ({}Hz, {}ch, {}-bit, {:.2} MB, {:.1} MB/s, {}s)",
|
"Player: Starting dynamic streaming for track {} ({}Hz, {}ch, {}-bit, {:.2} MB, {:.1} MB/s, {}s, start={}s)",
|
||||||
track_id,
|
track_id,
|
||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
bit_depth,
|
bit_depth,
|
||||||
content_length as f64 / (1024.0 * 1024.0),
|
content_length as f64 / (1024.0 * 1024.0),
|
||||||
speed_mbps,
|
speed_mbps,
|
||||||
duration_secs
|
duration_secs,
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
|
|
||||||
// Update shared state with actual stream quality
|
// Update shared state with actual stream quality
|
||||||
@@ -3383,6 +3478,8 @@ impl Player {
|
|||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
start_position_secs,
|
||||||
|
content_length,
|
||||||
})
|
})
|
||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
log::error!("Player: Failed to send streaming command: {}", e);
|
log::error!("Player: Failed to send streaming command: {}", e);
|
||||||
|
|||||||
@@ -652,6 +652,7 @@ async fn play_queue_track(ctx: &ApiContext, track: QueueTrack) -> Result<QueueTr
|
|||||||
None,
|
None,
|
||||||
None,
|
None,
|
||||||
Some(track.duration_secs),
|
Some(track.duration_secs),
|
||||||
|
None, // start_position_secs: remote-driven play always starts at 0
|
||||||
bridge,
|
bridge,
|
||||||
offline_cache,
|
offline_cache,
|
||||||
audio_settings,
|
audio_settings,
|
||||||
@@ -884,6 +885,7 @@ async fn play_queue_index(
|
|||||||
None,
|
None,
|
||||||
None,
|
None,
|
||||||
Some(t.duration_secs),
|
Some(t.duration_secs),
|
||||||
|
None, // start_position_secs: remote-driven play always starts at 0
|
||||||
bridge,
|
bridge,
|
||||||
offline_cache,
|
offline_cache,
|
||||||
audio_settings,
|
audio_settings,
|
||||||
@@ -1059,6 +1061,7 @@ async fn play_album(
|
|||||||
None,
|
None,
|
||||||
None,
|
None,
|
||||||
Some(first_track.duration_secs),
|
Some(first_track.duration_secs),
|
||||||
|
None, // start_position_secs: remote-driven play always starts at 0
|
||||||
bridge,
|
bridge,
|
||||||
offline_cache,
|
offline_cache,
|
||||||
audio_settings,
|
audio_settings,
|
||||||
|
|||||||
@@ -887,12 +887,18 @@ pub struct V2PlayTrackResult {
|
|||||||
/// 3. Downloads audio
|
/// 3. Downloads audio
|
||||||
/// 4. Plays via CoreBridge.player() (qbz-player crate)
|
/// 4. Plays via CoreBridge.player() (qbz-player crate)
|
||||||
/// 5. Caches for future playback
|
/// 5. Caches for future playback
|
||||||
|
// `start_position_secs` (#315): when > 0, the streaming path waits for
|
||||||
|
// enough buffer to cover this offset and pre-skips decoder output up to
|
||||||
|
// that point, then engages audio. Avoids the play-then-seek race that
|
||||||
|
// silently dropped the saved position on first-track session restore.
|
||||||
|
// 0 / None = start from beginning.
|
||||||
#[tauri::command]
|
#[tauri::command]
|
||||||
pub async fn v2_play_track(
|
pub async fn v2_play_track(
|
||||||
track_id: u64,
|
track_id: u64,
|
||||||
quality: Option<String>,
|
quality: Option<String>,
|
||||||
force_lowest_quality: Option<bool>,
|
force_lowest_quality: Option<bool>,
|
||||||
duration_secs: Option<u64>,
|
duration_secs: Option<u64>,
|
||||||
|
start_position_secs: Option<u64>,
|
||||||
bridge: State<'_, CoreBridgeState>,
|
bridge: State<'_, CoreBridgeState>,
|
||||||
offline_cache: State<'_, OfflineCacheState>,
|
offline_cache: State<'_, OfflineCacheState>,
|
||||||
audio_settings: State<'_, AudioSettingsState>,
|
audio_settings: State<'_, AudioSettingsState>,
|
||||||
@@ -901,6 +907,7 @@ pub async fn v2_play_track(
|
|||||||
app_handle: tauri::AppHandle,
|
app_handle: tauri::AppHandle,
|
||||||
runtime: State<'_, RuntimeManagerState>,
|
runtime: State<'_, RuntimeManagerState>,
|
||||||
) -> Result<V2PlayTrackResult, RuntimeError> {
|
) -> Result<V2PlayTrackResult, RuntimeError> {
|
||||||
|
let start_position_secs = start_position_secs.unwrap_or(0);
|
||||||
// Runtime contract: require CoreBridge auth for V2 playback
|
// Runtime contract: require CoreBridge auth for V2 playback
|
||||||
runtime
|
runtime
|
||||||
.manager()
|
.manager()
|
||||||
@@ -1294,6 +1301,7 @@ pub async fn v2_play_track(
|
|||||||
total_flac_size,
|
total_flac_size,
|
||||||
speed_mbps,
|
speed_mbps,
|
||||||
duration_secs.unwrap_or(0),
|
duration_secs.unwrap_or(0),
|
||||||
|
start_position_secs,
|
||||||
)
|
)
|
||||||
.map_err(RuntimeError::Internal)?;
|
.map_err(RuntimeError::Internal)?;
|
||||||
|
|
||||||
@@ -1528,6 +1536,7 @@ pub async fn v2_play_track(
|
|||||||
stream_info.content_length,
|
stream_info.content_length,
|
||||||
stream_info.speed_mbps,
|
stream_info.speed_mbps,
|
||||||
duration_secs.unwrap_or(0),
|
duration_secs.unwrap_or(0),
|
||||||
|
start_position_secs,
|
||||||
)
|
)
|
||||||
.map_err(RuntimeError::Internal)?;
|
.map_err(RuntimeError::Internal)?;
|
||||||
|
|
||||||
|
|||||||
+122
-18
@@ -86,6 +86,17 @@ enum AudioCommand {
|
|||||||
sample_rate: u32,
|
sample_rate: u32,
|
||||||
channels: u16,
|
channels: u16,
|
||||||
duration_secs: u64,
|
duration_secs: u64,
|
||||||
|
/// Resume offset in seconds. 0 = start from beginning. >0 = the
|
||||||
|
/// audio thread waits until enough bytes are buffered to cover
|
||||||
|
/// (position + headroom) seconds, then decodes-and-discards
|
||||||
|
/// samples up to `start_position_secs` before feeding the engine.
|
||||||
|
/// Avoids the play-then-seek race where Seek would silently
|
||||||
|
/// bail because `current_audio_data` is unset for streaming.
|
||||||
|
start_position_secs: u64,
|
||||||
|
/// Total content length in bytes. Used together with
|
||||||
|
/// `duration_secs` to estimate bytes-per-second when computing
|
||||||
|
/// how much buffer is needed for `start_position_secs > 0`.
|
||||||
|
content_length: u64,
|
||||||
},
|
},
|
||||||
/// Pause playback
|
/// Pause playback
|
||||||
Pause,
|
Pause,
|
||||||
@@ -1619,13 +1630,16 @@ impl Player {
|
|||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
start_position_secs,
|
||||||
|
content_length,
|
||||||
} => {
|
} => {
|
||||||
log::info!(
|
log::info!(
|
||||||
"Audio thread: starting streaming playback for track {} ({}Hz, {} channels, {}s)",
|
"Audio thread: starting streaming playback for track {} ({}Hz, {} channels, {}s, start={}s)",
|
||||||
track_id,
|
track_id,
|
||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs
|
duration_secs,
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
*pause_suspend_deadline = None;
|
*pause_suspend_deadline = None;
|
||||||
|
|
||||||
@@ -1839,12 +1853,45 @@ impl Player {
|
|||||||
let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0;
|
let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0;
|
||||||
apply_engine_volume(&stream_opt, &engine, volume);
|
apply_engine_volume(&stream_opt, &engine, volume);
|
||||||
|
|
||||||
// Wait for minimum buffer before starting playback
|
// Wait for minimum buffer before starting playback.
|
||||||
|
// When start_position_secs > 0 (session resume),
|
||||||
|
// also wait for enough buffer to cover the resume
|
||||||
|
// offset plus an 8s headroom — the eager pre-skip
|
||||||
|
// below will decode-and-discard up to the offset,
|
||||||
|
// and we need bytes in the buffer to feed it
|
||||||
|
// without blocking the audio device.
|
||||||
log::info!("Streaming: waiting for initial buffer...");
|
log::info!("Streaming: waiting for initial buffer...");
|
||||||
let start_wait = Instant::now();
|
let start_wait = Instant::now();
|
||||||
let max_wait = Duration::from_secs(30);
|
let max_wait = Duration::from_secs(60);
|
||||||
|
|
||||||
while !source.has_min_buffer() && start_wait.elapsed() < max_wait {
|
let bytes_per_sec_estimate: u64 = if duration_secs > 0
|
||||||
|
&& content_length > 0
|
||||||
|
{
|
||||||
|
content_length / duration_secs
|
||||||
|
} else {
|
||||||
|
// Conservative fallback: ~200 KB/s ≈ 16/44 FLAC.
|
||||||
|
200_000
|
||||||
|
};
|
||||||
|
let resume_buffer_target: u64 = if start_position_secs > 0 {
|
||||||
|
bytes_per_sec_estimate
|
||||||
|
.saturating_mul(start_position_secs.saturating_add(8))
|
||||||
|
} else {
|
||||||
|
0
|
||||||
|
};
|
||||||
|
|
||||||
|
let buffer_sufficient = |src: &Arc<BufferedMediaSource>| -> bool {
|
||||||
|
if !src.has_min_buffer() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if resume_buffer_target == 0 {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
(src.buffer_size() as u64) >= resume_buffer_target
|
||||||
|
};
|
||||||
|
|
||||||
|
while !buffer_sufficient(&source)
|
||||||
|
&& start_wait.elapsed() < max_wait
|
||||||
|
{
|
||||||
std::thread::sleep(Duration::from_millis(50));
|
std::thread::sleep(Duration::from_millis(50));
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1852,11 +1899,20 @@ impl Player {
|
|||||||
log::error!("Streaming: timeout waiting for initial buffer");
|
log::error!("Streaming: timeout waiting for initial buffer");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if resume_buffer_target > 0
|
||||||
|
&& (source.buffer_size() as u64) < resume_buffer_target
|
||||||
|
{
|
||||||
|
log::warn!(
|
||||||
|
"Streaming: timed out waiting for resume buffer (got {} bytes, wanted {}); pre-skip may underrun briefly",
|
||||||
|
source.buffer_size(),
|
||||||
|
resume_buffer_target
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
let buffer_wait_ms = start_wait.elapsed().as_millis();
|
let buffer_wait_ms = start_wait.elapsed().as_millis();
|
||||||
log::info!(
|
log::info!(
|
||||||
"Streaming: initial buffer ready in {}ms, creating incremental decoder...",
|
"Streaming: buffer ready in {}ms ({} bytes, target {}), creating incremental decoder...",
|
||||||
buffer_wait_ms
|
buffer_wait_ms, source.buffer_size(), resume_buffer_target
|
||||||
);
|
);
|
||||||
|
|
||||||
// Create incremental streaming source - this starts playback IMMEDIATELY
|
// Create incremental streaming source - this starts playback IMMEDIATELY
|
||||||
@@ -1934,8 +1990,41 @@ impl Player {
|
|||||||
thread_state.set_normalization_gain(normalization);
|
thread_state.set_normalization_gain(normalization);
|
||||||
|
|
||||||
// Box the incremental source to match the expected type
|
// Box the incremental source to match the expected type
|
||||||
let source_to_play: Box<dyn Source<Item = f32> + Send> =
|
let mut source_to_play: Box<dyn Source<Item = f32> + Send> =
|
||||||
Box::new(incremental_source);
|
Box::new(incremental_source);
|
||||||
|
|
||||||
|
// Eager pre-skip for session resume. We decode and
|
||||||
|
// discard samples here so that the engine's first
|
||||||
|
// pull doesn't have to do the work synchronously,
|
||||||
|
// which would underrun the audio device for
|
||||||
|
// multi-second offsets. The buffer wait above
|
||||||
|
// guarantees there's enough downloaded data to
|
||||||
|
// feed this loop without blocking on bytes.
|
||||||
|
if start_position_secs > 0 {
|
||||||
|
let target_samples: u64 = (start_position_secs)
|
||||||
|
.saturating_mul(actual_sr as u64)
|
||||||
|
.saturating_mul(actual_ch as u64);
|
||||||
|
let skip_start = Instant::now();
|
||||||
|
let mut skipped: u64 = 0;
|
||||||
|
while skipped < target_samples {
|
||||||
|
if source_to_play.next().is_none() {
|
||||||
|
log::warn!(
|
||||||
|
"Resume: source ended before reaching {}s (pre-skipped {} samples)",
|
||||||
|
start_position_secs,
|
||||||
|
skipped
|
||||||
|
);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
skipped += 1;
|
||||||
|
}
|
||||||
|
log::info!(
|
||||||
|
"Resume: pre-skipped {} samples ({}s) in {}ms",
|
||||||
|
skipped,
|
||||||
|
start_position_secs,
|
||||||
|
skip_start.elapsed().as_millis()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
// Wrap source with diagnostic, normalization, and visualizer
|
// Wrap source with diagnostic, normalization, and visualizer
|
||||||
let source_to_play = wrap_source(
|
let source_to_play = wrap_source(
|
||||||
source_to_play,
|
source_to_play,
|
||||||
@@ -1950,16 +2039,19 @@ impl Player {
|
|||||||
}
|
}
|
||||||
|
|
||||||
thread_state.is_playing.store(true, Ordering::SeqCst);
|
thread_state.is_playing.store(true, Ordering::SeqCst);
|
||||||
thread_state.position.store(0, Ordering::SeqCst);
|
thread_state
|
||||||
|
.position
|
||||||
|
.store(start_position_secs, Ordering::SeqCst);
|
||||||
thread_state
|
thread_state
|
||||||
.current_track_id
|
.current_track_id
|
||||||
.store(track_id, Ordering::SeqCst);
|
.store(track_id, Ordering::SeqCst);
|
||||||
thread_state.start_playback_timer(0);
|
thread_state.start_playback_timer(start_position_secs);
|
||||||
|
|
||||||
*current_engine = Some(engine);
|
*current_engine = Some(engine);
|
||||||
log::info!(
|
log::info!(
|
||||||
"Audio thread: streaming playback STARTED in {}ms (incremental decode active)",
|
"Audio thread: streaming playback STARTED in {}ms at {}s (incremental decode active)",
|
||||||
start_wait.elapsed().as_millis()
|
start_wait.elapsed().as_millis(),
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
AudioCommand::Pause => {
|
AudioCommand::Pause => {
|
||||||
@@ -2689,7 +2781,10 @@ impl Player {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Play from streaming source (starts playback before full download)
|
/// Play from streaming source (starts playback before full download)
|
||||||
/// Returns the BufferWriter so caller can push data as it downloads
|
/// Returns the BufferWriter so caller can push data as it downloads.
|
||||||
|
/// `start_position_secs` > 0 turns this into a session-resume play:
|
||||||
|
/// the audio thread waits for enough buffer to cover the offset and
|
||||||
|
/// pre-skips decode output up to that point before engaging audio.
|
||||||
pub fn play_streaming(
|
pub fn play_streaming(
|
||||||
&self,
|
&self,
|
||||||
track_id: u64,
|
track_id: u64,
|
||||||
@@ -2698,14 +2793,16 @@ impl Player {
|
|||||||
content_length: u64,
|
content_length: u64,
|
||||||
buffer_seconds: u8,
|
buffer_seconds: u8,
|
||||||
duration_secs: u64,
|
duration_secs: u64,
|
||||||
|
start_position_secs: u64,
|
||||||
) -> Result<BufferWriter, String> {
|
) -> Result<BufferWriter, String> {
|
||||||
log::info!(
|
log::info!(
|
||||||
"Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total, {}s)",
|
"Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total, {}s, start={}s)",
|
||||||
track_id,
|
track_id,
|
||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
content_length,
|
content_length,
|
||||||
duration_secs
|
duration_secs,
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
|
|
||||||
// Use StreamingConfig::from_seconds for proper buffer sizing
|
// Use StreamingConfig::from_seconds for proper buffer sizing
|
||||||
@@ -2721,6 +2818,8 @@ impl Player {
|
|||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
start_position_secs,
|
||||||
|
content_length,
|
||||||
})
|
})
|
||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
log::error!("Player: Failed to send streaming command: {}", e);
|
log::error!("Player: Failed to send streaming command: {}", e);
|
||||||
@@ -2732,7 +2831,8 @@ impl Player {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Play from streaming source with dynamic buffer based on measured speed
|
/// Play from streaming source with dynamic buffer based on measured speed
|
||||||
/// Returns the BufferWriter so caller can push data as it downloads
|
/// Returns the BufferWriter so caller can push data as it downloads.
|
||||||
|
/// `start_position_secs` > 0 signals session resume (see `play_streaming`).
|
||||||
pub fn play_streaming_dynamic(
|
pub fn play_streaming_dynamic(
|
||||||
&self,
|
&self,
|
||||||
track_id: u64,
|
track_id: u64,
|
||||||
@@ -2742,16 +2842,18 @@ impl Player {
|
|||||||
content_length: u64,
|
content_length: u64,
|
||||||
speed_mbps: f64,
|
speed_mbps: f64,
|
||||||
duration_secs: u64,
|
duration_secs: u64,
|
||||||
|
start_position_secs: u64,
|
||||||
) -> Result<BufferWriter, String> {
|
) -> Result<BufferWriter, String> {
|
||||||
log::info!(
|
log::info!(
|
||||||
"Player: Starting dynamic streaming for track {} ({}Hz, {}ch, {}-bit, {:.2} MB, {:.1} MB/s, {}s)",
|
"Player: Starting dynamic streaming for track {} ({}Hz, {}ch, {}-bit, {:.2} MB, {:.1} MB/s, {}s, start={}s)",
|
||||||
track_id,
|
track_id,
|
||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
bit_depth,
|
bit_depth,
|
||||||
content_length as f64 / (1024.0 * 1024.0),
|
content_length as f64 / (1024.0 * 1024.0),
|
||||||
speed_mbps,
|
speed_mbps,
|
||||||
duration_secs
|
duration_secs,
|
||||||
|
start_position_secs
|
||||||
);
|
);
|
||||||
|
|
||||||
// Update shared state with actual stream quality
|
// Update shared state with actual stream quality
|
||||||
@@ -2770,6 +2872,8 @@ impl Player {
|
|||||||
sample_rate,
|
sample_rate,
|
||||||
channels,
|
channels,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
start_position_secs,
|
||||||
|
content_length,
|
||||||
})
|
})
|
||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
log::error!("Player: Failed to send streaming command: {}", e);
|
log::error!("Player: Failed to send streaming command: {}", e);
|
||||||
|
|||||||
@@ -131,6 +131,7 @@ async fn stream_remote_track_into_player(
|
|||||||
stream_info.content_length,
|
stream_info.content_length,
|
||||||
stream_info.speed_mbps,
|
stream_info.speed_mbps,
|
||||||
duration_secs,
|
duration_secs,
|
||||||
|
0, // QConnect tracks always start from 0; resume is local-only
|
||||||
)
|
)
|
||||||
.map_err(|err| format!("start streaming remote track {track_id}: {err}"))?;
|
.map_err(|err| format!("start streaming remote track {track_id}: {err}"))?;
|
||||||
|
|
||||||
|
|||||||
@@ -105,21 +105,27 @@ export async function cmdPlayTrack(
|
|||||||
quality?: string,
|
quality?: string,
|
||||||
durationSecs?: number | null,
|
durationSecs?: number | null,
|
||||||
forceLowestQuality?: boolean | null,
|
forceLowestQuality?: boolean | null,
|
||||||
|
startPositionSecs?: number | null,
|
||||||
): Promise<unknown> {
|
): Promise<unknown> {
|
||||||
const target = getTarget();
|
const target = getTarget();
|
||||||
if (target.type === 'qbzd') {
|
if (target.type === 'qbzd') {
|
||||||
return remotePost('/api/playback/play-track', { track_id: trackId, quality });
|
return remotePost('/api/playback/play-track', { track_id: trackId, quality });
|
||||||
} else {
|
} else {
|
||||||
// Parameter names MUST match Tauri's camelCase mapping of Rust
|
// Parameter names MUST match Tauri's camelCase mapping of Rust
|
||||||
// v2_play_track(track_id, quality, force_lowest_quality, duration_secs).
|
// v2_play_track(track_id, quality, force_lowest_quality, duration_secs,
|
||||||
// duration_secs is required on the streaming path — without it the
|
// start_position_secs). duration_secs is required on the streaming
|
||||||
// backend stores duration=0 and current_position() clamps to 0,
|
// path — without it the backend stores duration=0 and
|
||||||
// freezing the seekbar (seen on session-restore first play).
|
// current_position() clamps to 0, freezing the seekbar (seen on
|
||||||
|
// session-restore first play). start_position_secs > 0 turns the
|
||||||
|
// call into a session-resume play (#315): backend buffers enough
|
||||||
|
// bytes to cover the offset and pre-skips decoder output before
|
||||||
|
// engaging audio, avoiding the play-then-seek race.
|
||||||
return invoke('v2_play_track', {
|
return invoke('v2_play_track', {
|
||||||
trackId,
|
trackId,
|
||||||
quality: quality ?? null,
|
quality: quality ?? null,
|
||||||
forceLowestQuality: forceLowestQuality ?? null,
|
forceLowestQuality: forceLowestQuality ?? null,
|
||||||
durationSecs: durationSecs ?? null,
|
durationSecs: durationSecs ?? null,
|
||||||
|
startPositionSecs: startPositionSecs ?? null,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -155,31 +155,53 @@ export async function clearSession(): Promise<void> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Debounce helper for position saves
|
// Throttle helper for position saves. We THROTTLE (not debounce) because
|
||||||
|
// position updates arrive continuously while a track plays — a true
|
||||||
|
// debounce would reset the 5s timer on every update and the save would
|
||||||
|
// never fire mid-playback, leaving the persisted position frozen at
|
||||||
|
// whatever value got written during the first quiet window after track
|
||||||
|
// start. (Real-world symptom: resume position always restored at ~2s,
|
||||||
|
// no matter how far the user played.) Throttling fires the first call
|
||||||
|
// immediately, then once per 5s window thereafter, with a tail call to
|
||||||
|
// flush the latest position when the window ends.
|
||||||
let positionSaveTimeout: ReturnType<typeof setTimeout> | null = null;
|
let positionSaveTimeout: ReturnType<typeof setTimeout> | null = null;
|
||||||
const POSITION_SAVE_DEBOUNCE_MS = 5000; // Save position every 5 seconds max
|
let lastSavedPositionMs = 0;
|
||||||
|
let lastQueuedPositionSecs = 0;
|
||||||
|
const POSITION_SAVE_THROTTLE_MS = 5000;
|
||||||
|
|
||||||
/**
|
|
||||||
* Debounced position save - call frequently, saves every 5 seconds
|
|
||||||
*/
|
|
||||||
export function debouncedSavePosition(positionSecs: number): void {
|
export function debouncedSavePosition(positionSecs: number): void {
|
||||||
if (positionSaveTimeout) {
|
lastQueuedPositionSecs = positionSecs;
|
||||||
clearTimeout(positionSaveTimeout);
|
const now = Date.now();
|
||||||
}
|
const elapsed = now - lastSavedPositionMs;
|
||||||
positionSaveTimeout = setTimeout(() => {
|
if (elapsed >= POSITION_SAVE_THROTTLE_MS) {
|
||||||
|
lastSavedPositionMs = now;
|
||||||
saveSessionPosition(positionSecs);
|
saveSessionPosition(positionSecs);
|
||||||
|
if (positionSaveTimeout) {
|
||||||
|
clearTimeout(positionSaveTimeout);
|
||||||
|
positionSaveTimeout = null;
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (positionSaveTimeout) return;
|
||||||
|
positionSaveTimeout = setTimeout(() => {
|
||||||
positionSaveTimeout = null;
|
positionSaveTimeout = null;
|
||||||
}, POSITION_SAVE_DEBOUNCE_MS);
|
lastSavedPositionMs = Date.now();
|
||||||
|
saveSessionPosition(lastQueuedPositionSecs);
|
||||||
|
}, POSITION_SAVE_THROTTLE_MS - elapsed);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Force save position immediately (e.g., on pause or app close)
|
* Force save position immediately (e.g., on pause or app close).
|
||||||
|
* Resets throttle bookkeeping so the next debouncedSavePosition fires
|
||||||
|
* fresh.
|
||||||
*/
|
*/
|
||||||
export function flushPositionSave(positionSecs: number): void {
|
export function flushPositionSave(positionSecs: number): void {
|
||||||
if (positionSaveTimeout) {
|
if (positionSaveTimeout) {
|
||||||
clearTimeout(positionSaveTimeout);
|
clearTimeout(positionSaveTimeout);
|
||||||
positionSaveTimeout = null;
|
positionSaveTimeout = null;
|
||||||
}
|
}
|
||||||
|
lastSavedPositionMs = Date.now();
|
||||||
|
lastQueuedPositionSecs = positionSecs;
|
||||||
saveSessionPosition(positionSecs);
|
saveSessionPosition(positionSecs);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -441,6 +441,16 @@ export function setQueueEnded(ended: boolean): void {
|
|||||||
*/
|
*/
|
||||||
export function setPendingSessionRestore(trackId: number, positionSecs?: number): void {
|
export function setPendingSessionRestore(trackId: number, positionSecs?: number): void {
|
||||||
pendingSessionRestore = { trackId, positionSecs };
|
pendingSessionRestore = { trackId, positionSecs };
|
||||||
|
// Pre-position the seekbar at the saved offset so the visual state
|
||||||
|
// reflects "paused at X" from the moment the app opens, instead of
|
||||||
|
// sitting at 0:00 until the user hits play. The backend hasn't
|
||||||
|
// loaded the track yet, but currentTime is the source for the
|
||||||
|
// seekbar render — when togglePlay eventually fires it'll already be
|
||||||
|
// anchored to the same position the backend resumes at.
|
||||||
|
if (positionSecs && positionSecs > 0) {
|
||||||
|
currentTime = positionSecs;
|
||||||
|
notifyListeners();
|
||||||
|
}
|
||||||
console.log(
|
console.log(
|
||||||
'[Player] Set pending session restore:',
|
'[Player] Set pending session restore:',
|
||||||
trackId,
|
trackId,
|
||||||
@@ -531,32 +541,45 @@ export async function togglePlay(): Promise<void> {
|
|||||||
if (result.bit_depth && result.bit_depth > 0) {
|
if (result.bit_depth && result.bit_depth > 0) {
|
||||||
currentTrack.bitDepth = result.bit_depth;
|
currentTrack.bitDepth = result.bit_depth;
|
||||||
}
|
}
|
||||||
|
// Plex loads via Symphonia just like a local file — Seek
|
||||||
|
// works once the engine is up.
|
||||||
|
if (restorePosition > 1) {
|
||||||
|
pendingSeekPosition = restorePosition;
|
||||||
|
void flushPendingSeek();
|
||||||
|
}
|
||||||
} else if (currentTrack.isLocal || currentTrack.id < 0) {
|
} else if (currentTrack.isLocal || currentTrack.id < 0) {
|
||||||
// Local filesystem track
|
// Local filesystem track. Apply the user's saved playback
|
||||||
|
// position (#317 opt-in) via the seek path — local tracks
|
||||||
|
// load full audio data into the player, so Seek works as soon
|
||||||
|
// as the engine is up.
|
||||||
const localTrackId = Math.abs(currentTrack.id);
|
const localTrackId = Math.abs(currentTrack.id);
|
||||||
await invoke('v2_library_play_track', { trackId: localTrackId });
|
await invoke('v2_library_play_track', { trackId: localTrackId });
|
||||||
|
if (restorePosition > 1) {
|
||||||
|
pendingSeekPosition = restorePosition;
|
||||||
|
void flushPendingSeek();
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
// Qobuz track - use v2_play_track. Pass duration so the
|
// Qobuz track — use v2_play_track. Pass duration so the
|
||||||
// streaming backend's current_position() doesn't clamp to 0
|
// streaming backend's current_position() doesn't clamp to 0
|
||||||
// (the value flows into thread_state.duration and seekbar
|
// (the value flows into thread_state.duration and seekbar
|
||||||
// progress is capped by .min(duration)).
|
// progress is capped by .min(duration)). For session resume,
|
||||||
|
// hand the restore offset directly to v2_play_track via
|
||||||
|
// startPositionSecs (#315 extension): the streaming path
|
||||||
|
// waits for enough buffer to cover the offset and pre-skips
|
||||||
|
// decoder output before engaging audio, instead of the
|
||||||
|
// play-then-seek dance that silently dropped the seek
|
||||||
|
// because the audio thread's Seek handler bails when there
|
||||||
|
// is no `current_audio_data` (streaming uses
|
||||||
|
// `current_streaming_source` instead).
|
||||||
await cmdPlayTrack(
|
await cmdPlayTrack(
|
||||||
currentTrack.id,
|
currentTrack.id,
|
||||||
getStreamingQuality(),
|
getStreamingQuality(),
|
||||||
currentTrack.duration ? Math.round(currentTrack.duration) : null,
|
currentTrack.duration ? Math.round(currentTrack.duration) : null,
|
||||||
|
null,
|
||||||
|
restorePosition > 1 ? restorePosition : null,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Apply the user's saved playback position (#317 opt-in).
|
|
||||||
// Queue via the existing pendingSeekPosition / flushPendingSeek
|
|
||||||
// path so the seek runs against an already-loaded stream and
|
|
||||||
// honors the in-flight guard. If the position is at or near 0,
|
|
||||||
// skip — the load already starts there.
|
|
||||||
if (restorePosition > 1) {
|
|
||||||
pendingSeekPosition = restorePosition;
|
|
||||||
void flushPendingSeek();
|
|
||||||
}
|
|
||||||
|
|
||||||
} else {
|
} else {
|
||||||
await cmdResume();
|
await cmdResume();
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user