Fix/resolve probe tasks (#644)

* Fixed issues with probe and resolve tasks
* hls and catchup session fixes
* fixed problem with grace and shared stream
- added fix for missing episode title

backend/src/processing/processor/library.rs

@@ -3,7 +3,7 @@ use crate::model::{AppConfig, ConfigInput};
 use shared::concat_string;
 use shared::error::TuliproxError;
 use shared::model::UUIDType;
-use shared::model::{EpisodeStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamDetailSeasonProperties, SeriesStreamProperties, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster};
+use shared::model::{EpisodeStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamDetailSeasonProperties, SeriesStreamProperties, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster, normalize_episode_title};
 use shared::utils::{generate_playlist_uuid, Internable};
 use std::collections::HashMap;
 use std::path::Path;
@@ -263,6 +263,8 @@ pub fn metadata_cache_entry_to_xtream_series_info(
                 .unwrap_or_default();
             let episode_release_date = episode.aired.as_ref().map(ToString::to_string).unwrap_or_default();
             let tmdb_id = (episode.tmdb_id > 0).then_some(episode.tmdb_id);
+            let raw_episode_title: Arc<str> = episode.title.clone().into();
+            let series_title: Arc<str> = series.title.clone().into();
 
             let season_entry = season_data.entry(episode.season).or_insert_with(|| {
                 SeriesStreamDetailSeasonProperties {
@@ -283,7 +285,7 @@ pub fn metadata_cache_entry_to_xtream_series_info(
                 id: tmdb_id.unwrap_or_default(),
                 episode_num: episode.episode,
                 season: episode.season,
-                title: episode.title.clone().into(),
+                title: normalize_episode_title(&raw_episode_title, &series_title, episode.season, episode.episode),
                 container_extension: container_extension.into(),
                 custom_sid: None,
                 added: episode.file_modified.to_string().into(),
@@ -344,4 +346,4 @@ pub fn metadata_cache_entry_to_xtream_series_info(
     };
 
     Some(StreamProperties::Series(Box::new(properties)))
-}
+}

backend/src/processing/processor/mod.rs

@@ -9,43 +9,19 @@ mod trakt;
 mod library;
 mod stream_probe;
 mod filtered_playlist_source;
+mod probe_handle_guard;
+mod resolve_options;
 pub use self::playlist::*;
 pub use self::epg::*;
 pub use self::xtream::*;
 pub use self::xtream_vod::*;
 pub use self::xtream_series::*;
 pub use self::stream_probe::*;
+pub(crate) use self::probe_handle_guard::*;
+pub use self::resolve_options::*;
 use crate::api::model::ProviderHandle;
-use shared::create_bitset;
 use tokio_util::sync::CancellationToken;
 
-create_bitset!(u8, ResolveOptionsFlags, Resolve, TmdbMissing, Probe, Background);
-
-pub struct ResolveOptions {
-    flags: ResolveOptionsFlagsSet,
-    pub resolve_delay: u16,
-}
-
-impl ResolveOptions {
-    #[inline]
-    pub fn has_flag(&self, flag: ResolveOptionsFlags) -> bool {
-        self.flags.contains(flag)
-    }
-    #[inline]
-    pub fn unset_flag(&mut self, flag: ResolveOptionsFlags) {
-        self.flags.unset(flag);
-    }
-}
-
-impl Default for ResolveOptions {
-    fn default() -> Self {
-        Self {
-            flags: ResolveOptionsFlagsSet::from_variants(&[ResolveOptionsFlags::Background]),
-            resolve_delay: default_resolve_delay_secs(),
-        }
-    }
-}
-
 pub(crate) const FOREGROUND_BATCH_SIZE: usize = 200;
 pub(crate) const FOREGROUND_RETRY_BATCH_MAX_SIZE: usize = FOREGROUND_BATCH_SIZE * 4;
 pub(crate) const FOREGROUND_MIN_RETRY_DELAY_SECS: u64 = 1;
@@ -293,4 +269,3 @@ macro_rules! process_foreground_retry_once {
     };
 }
 pub(crate) use process_foreground_retry_once;
-use shared::utils::default_resolve_delay_secs;

backend/src/processing/processor/playlist.rs

@@ -1180,7 +1180,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
                                     delay: probe_delay,
                                     interval: interval_secs,
                                 };
-                                if mgr.is_redundant_with_pending_task(input_name.as_ref(), &task) {
+                                if mgr.should_skip_enqueue(input_name.clone(), &task).await {
                                     continue;
                                 }
                                 if log_enabled!(Level::Debug) {
@@ -1249,7 +1249,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
             reason: ResolveReasonSet::from_variants(&[ResolveReason::MissingDetails]),
             delay: opts.probe_delay,
         };
-        if mgr.is_redundant_with_pending_task(input_name.as_ref(), &task) {
+        if mgr.should_skip_enqueue(input_name.clone(), &task).await {
             continue;
         }
         debug!(

backend/src/processing/processor/probe_handle_guard.rs

@@ -0,0 +1,120 @@
+use crate::api::model::{ActiveProviderManager, ProviderHandle};
+use std::sync::Arc;
+
+pub(crate) struct ProbeHandleGuard {
+    manager: Arc<ActiveProviderManager>,
+    handle: Option<ProviderHandle>,
+}
+
+impl ProbeHandleGuard {
+    pub(crate) fn new(manager: &Arc<ActiveProviderManager>, handle: ProviderHandle) -> Self {
+        Self {
+            manager: Arc::clone(manager),
+            handle: Some(handle),
+        }
+    }
+
+    #[inline]
+    pub(crate) fn handle(&self) -> Option<&ProviderHandle> { self.handle.as_ref() }
+
+    pub(crate) async fn release(mut self) {
+        if let Some(handle) = self.handle.take() {
+            self.manager.release_handle(&handle).await;
+        }
+    }
+}
+
+impl Drop for ProbeHandleGuard {
+    fn drop(&mut self) {
+        let Some(handle) = self.handle.take() else {
+            return;
+        };
+        let manager = Arc::clone(&self.manager);
+        if let Ok(runtime_handle) = tokio::runtime::Handle::try_current() {
+            runtime_handle.spawn(async move {
+                manager.release_handle(&handle).await;
+            });
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ProbeHandleGuard;
+    use crate::{
+        api::model::{ActiveProviderManager, EventManager},
+        model::{AppConfig, Config, ConfigInput, SourcesConfig},
+        utils::FileLockManager,
+    };
+    use arc_swap::{ArcSwap, ArcSwapOption};
+    use shared::{
+        model::{ConfigPaths, InputFetchMethod, InputType},
+        utils::{default_probe_user_priority, Internable},
+    };
+    use std::{collections::HashMap, sync::Arc};
+
+    fn create_test_app_config() -> AppConfig {
+        let input = Arc::new(ConfigInput {
+            id: 1,
+            name: "provider_1".intern(),
+            input_type: InputType::Xtream,
+            headers: HashMap::default(),
+            url: "http://provider-1.example".to_string(),
+            username: Some("user1".to_string()),
+            password: Some("pass1".to_string()),
+            enabled: true,
+            priority: 0,
+            max_connections: 1,
+            method: InputFetchMethod::default(),
+            ..ConfigInput::default()
+        });
+
+        let sources = SourcesConfig { inputs: vec![input], ..SourcesConfig::default() };
+
+        AppConfig {
+            config: Arc::new(ArcSwap::from_pointee(Config::default())),
+            sources: Arc::new(ArcSwap::from_pointee(sources)),
+            hdhomerun: Arc::new(ArcSwapOption::default()),
+            api_proxy: Arc::new(ArcSwapOption::default()),
+            file_locks: Arc::new(FileLockManager::default()),
+            paths: Arc::new(ArcSwap::from_pointee(ConfigPaths {
+                home_path: String::new(),
+                config_path: String::new(),
+                storage_path: String::new(),
+                config_file_path: String::new(),
+                sources_file_path: String::new(),
+                mapping_file_path: None,
+                mapping_files_used: None,
+                template_file_path: None,
+                template_files_used: None,
+                api_proxy_file_path: String::new(),
+                custom_stream_response_path: None,
+            })),
+            custom_stream_response: Arc::new(ArcSwapOption::default()),
+            access_token_secret: [0; 32],
+            encrypt_secret: [0; 16],
+            ffprobe_available: Arc::default(),
+        }
+    }
+
+    #[tokio::test]
+    async fn probe_handle_guard_releases_provider_slot_on_drop() {
+        let app_cfg = create_test_app_config();
+        let event_manager = Arc::new(EventManager::new());
+        let manager = Arc::new(ActiveProviderManager::new(&app_cfg, &event_manager));
+        let input_name = "provider_1".intern();
+
+        let handle = manager
+            .acquire_connection_for_probe(&input_name, default_probe_user_priority())
+            .await
+            .expect("probe allocation should succeed");
+        assert_eq!(manager.get_provider_connections_count().await, 1);
+
+        let guard = ProbeHandleGuard::new(&manager, handle);
+        drop(guard);
+        tokio::task::yield_now().await;
+        tokio::task::yield_now().await;
+
+        assert_eq!(manager.get_provider_connections_count().await, 0);
+    }
+}

backend/src/processing/processor/resolve_options.rs

@@ -0,0 +1,29 @@
+use shared::create_bitset;
+
+create_bitset!(u8, ResolveOptionsFlags, Resolve, TmdbMissing, Probe, Background);
+
+pub struct ResolveOptions {
+    pub(crate) flags: ResolveOptionsFlagsSet,
+    pub resolve_delay: u16,
+}
+
+impl ResolveOptions {
+    #[inline]
+    pub fn has_flag(&self, flag: ResolveOptionsFlags) -> bool {
+        self.flags.contains(flag)
+    }
+
+    #[inline]
+    pub fn unset_flag(&mut self, flag: ResolveOptionsFlags) {
+        self.flags.unset(flag);
+    }
+}
+
+impl Default for ResolveOptions {
+    fn default() -> Self {
+        Self {
+            flags: ResolveOptionsFlagsSet::from_variants(&[ResolveOptionsFlags::Background]),
+            resolve_delay: shared::utils::default_resolve_delay_secs(),
+        }
+    }
+}

backend/src/processing/processor/stream_probe.rs

@@ -1,7 +1,7 @@
 use crate::api::model::ActiveProviderManager;
 use crate::model::ConfigInput;
 use crate::model::{AppConfig};
-use crate::processing::processor::select_cancel_token;
+use crate::processing::processor::{select_cancel_token, ProbeHandleGuard};
 use crate::repository::{get_input_m3u_playlist_file_path, get_input_storage_path, get_input_local_library_playlist_file_path, xtream_get_file_path, BPlusTreeUpdate};
 use crate::utils::{debug_if_enabled, ffmpeg};
 use crate::utils::ffmpeg::{ProbeFailureKind, ProbeUrlOutcome};
@@ -90,7 +90,10 @@ pub async fn update_generic_stream_metadata(
     let acquired_handle = if !needs_provider_connection || active_handle.is_some() {
         None
     } else {
-        active_provider.acquire_connection_for_probe(&input.name, probe_priority).await
+        active_provider
+            .acquire_connection_for_probe(&input.name, probe_priority)
+            .await
+            .map(|handle| ProbeHandleGuard::new(active_provider, handle))
     };
 
     if needs_provider_connection && active_handle.is_none() && acquired_handle.is_none() {
@@ -117,7 +120,10 @@ pub async fn update_generic_stream_metadata(
 
     debug_if_enabled!("Probing Generic Stream '{unique_id}'");
 
-    let cancel_token = select_cancel_token(acquired_handle.as_ref(), active_handle);
+    let cancel_token = select_cancel_token(
+        acquired_handle.as_ref().and_then(ProbeHandleGuard::handle),
+        active_handle,
+    );
     let probe_data = ffmpeg::probe_url_with_cancel(
         &probe_url,
         user_agent.as_deref(),
@@ -130,7 +136,7 @@ pub async fn update_generic_stream_metadata(
     .await;
 
     if let Some(handle) = acquired_handle {
-        active_provider.release_handle(&handle).await;
+        handle.release().await;
     }
 
     let (raw_video, raw_audio) = match probe_data {

backend/src/processing/processor/xtream_series.rs

@@ -9,6 +9,7 @@ use crate::processing::parser::xtream::parse_xtream_series_info;
 use crate::processing::processor::playlist::{PlaylistProcessingContext, ProcessingPipe};
 use crate::processing::processor::{
     create_resolve_options_function_for_xtream_target, process_foreground_retry_once, select_cancel_token,
+    ProbeHandleGuard,
     ResolveOptions, ResolveOptionsFlags, FOREGROUND_BATCH_SIZE as BATCH_SIZE, FOREGROUND_MIN_RETRY_DELAY_SECS,
     FOREGROUND_RETRY_BATCH_MAX_SIZE as RETRY_BATCH_MAX_SIZE,
 };
@@ -111,7 +112,7 @@ async fn playlist_resolve_series_info(
     let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb);
 
     let groups_to_add = if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() {
-        queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled)
+        queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await
     } else {
         process_immediate_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await
     };
@@ -177,7 +178,7 @@ fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, proce
     }
 }
 
-fn queue_background_series_info(
+async fn queue_background_series_info(
     ctx: &PlaylistProcessingContext,
     fpl: &mut FetchedPlaylist<'_>,
     filter: impl Fn(&PlaylistItem) -> bool,
@@ -213,7 +214,7 @@ fn queue_background_series_info(
                     reason: reasons,
                     delay: resolve_options.resolve_delay,
                 };
-                if !mgr.is_redundant_with_pending_task(input_name_arc.as_ref(), &task) {
+                if !mgr.should_skip_enqueue(input_name_arc.clone(), &task).await {
                     if log_enabled!(Level::Debug) {
                         let has_details = pli.has_details();
                         let (has_tmdb, has_date) = match pli.header.additional_properties.as_ref() {
@@ -542,6 +543,7 @@ async fn update_series_info_immediate(
         probe_settings,
         db_query,
         None,
+        None,
     )
     .await
 }
@@ -639,6 +641,7 @@ pub async fn update_series_metadata(
     probe_settings: SeriesProbeSettings,
     db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
     tmdb_and_date_present_out: Option<&AtomicBool>,
+    probe_pending_out: Option<&AtomicBool>,
 ) -> Result<Option<SeriesStreamProperties>, TuliproxError> {
     let storage_dir = &app_config.config.load().storage_dir;
     let storage_path = get_input_storage_path(&input.name, storage_dir)
@@ -716,6 +719,7 @@ pub async fn update_series_metadata(
     let mut fetched_new = false;
     let mut properties_updated = false;
     let mut probe_failure: Option<ProbeFailureKind> = None;
+    let mut probe_pending = false;
 
     let display_id = series_id_opt.map_or_else(|| "StringID".to_string(), |v| v.to_string());
 
@@ -870,7 +874,10 @@ pub async fn update_series_metadata(
                         let temp_handle = if active_handle.is_some() {
                             None
                         } else {
-                            active_provider.acquire_connection_for_probe(&input.name, probe_priority).await
+                            active_provider
+                                .acquire_connection_for_probe(&input.name, probe_priority)
+                                .await
+                                .map(|handle| ProbeHandleGuard::new(active_provider, handle))
                         };
 
                         if active_handle.is_some() || temp_handle.is_some() {
@@ -890,7 +897,10 @@ pub async fn update_series_metadata(
                                 ep.title, ep.season, ep.episode_num, missing_reason
                             );
 
-                            let cancel_token = select_cancel_token(temp_handle.as_ref(), active_handle);
+                            let cancel_token = select_cancel_token(
+                                temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
+                                active_handle,
+                            );
                             match crate::utils::ffmpeg::probe_url_with_cancel(
                                 &episode_url,
                                 user_agent.as_deref(),
@@ -917,23 +927,26 @@ pub async fn update_series_metadata(
                                     probe_failure = Some(ProbeFailureKind::NotFound);
                                 }
                                 ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => {
+                                    probe_pending = true;
                                     if probe_failure.is_none() {
                                         probe_failure = Some(ProbeFailureKind::Other);
                                     }
                                 }
                                 ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => {
+                                    probe_pending = true;
                                     if probe_failure.is_none() {
                                         probe_failure = Some(ProbeFailureKind::Cancelled);
                                     }
                                 }
                             }
-
-                            if let Some(h) = temp_handle {
-                                active_provider.release_handle(&h).await;
-                            }
                         } else {
+                            probe_pending = true;
                             warn!("Skipping probe for series episode {} due to connection limits", ep.title);
                         }
+
+                        if let Some(guard) = temp_handle {
+                            guard.release().await;
+                        }
                     }
                 }
                 if probed_count > 0 {
@@ -949,6 +962,10 @@ pub async fn update_series_metadata(
         }
     }
 
+    if let Some(out) = probe_pending_out {
+        out.store(probe_pending, Ordering::Relaxed);
+    }
+
     let probe_only_unresolved = do_probe && !fetch_info && !resolve_tmdb && !properties_updated && !fetched_new;
     if probe_only_unresolved {
         if let Some(kind) = probe_failure {

backend/src/processing/processor/xtream_vod.rs

@@ -9,6 +9,7 @@ use crate::processing::input_cache::resolve_input_storage_path;
 use crate::processing::processor::playlist::PlaylistProcessingContext;
 use crate::processing::processor::{
     create_resolve_options_function_for_xtream_target, process_foreground_retry_once, select_cancel_token,
+    ProbeHandleGuard,
     ResolveOptions, ResolveOptionsFlags, FOREGROUND_BATCH_SIZE as BATCH_SIZE, FOREGROUND_MIN_RETRY_DELAY_SECS,
     FOREGROUND_RETRY_BATCH_MAX_SIZE as RETRY_BATCH_MAX_SIZE,
 };
@@ -133,7 +134,7 @@ async fn playlist_resolve_vod_info(
     let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb);
 
     if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() {
-        queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled);
+        queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled).await;
     } else {
         process_immediate_vod_info(ctx, fpl, filter, resolve_options, do_probe, resolve_tmdb_enabled).await;
     }
@@ -415,7 +416,7 @@ fn check_resolve_tmdb(
     }
 }
 
-fn queue_background_vod_info(
+async fn queue_background_vod_info(
     ctx: &PlaylistProcessingContext,
     fpl: &mut FetchedPlaylist<'_>,
     filter: impl Fn(&PlaylistItem) -> bool,
@@ -444,7 +445,7 @@ fn queue_background_vod_info(
         if !reasons.is_empty() {
             let task =
                 UpdateTask::ResolveVod { id: provider_id.clone(), reason: reasons, delay: resolve_options.resolve_delay };
-            if mgr.is_redundant_with_pending_task(input.name.as_ref(), &task) {
+            if mgr.should_skip_enqueue(input.name.clone(), &task).await {
                 continue;
             }
 
@@ -500,6 +501,7 @@ async fn update_vod_info_immediate(
         reasons.contains(ResolveReason::Probe),
         db_query,
         None,
+        None,
     )
     .await
 }
@@ -527,6 +529,7 @@ pub async fn update_vod_metadata(
     do_probe: bool,
     db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
     tmdb_resolved_out: Option<&AtomicBool>,
+    probe_pending_out: Option<&AtomicBool>,
 ) -> Result<Option<VideoStreamProperties>, TuliproxError> {
     let storage_dir = &app_config.config.load().storage_dir;
     let storage_path = resolve_input_storage_path(storage_dir, &input.name).await;
@@ -599,6 +602,7 @@ pub async fn update_vod_metadata(
     let mut fetched_new = false;
     let mut properties_updated = false;
     let mut probe_failure: Option<ProbeFailureKind> = None;
+    let mut probe_pending = false;
 
     // Determine the title to use for logging and fallback.
     // Cloning here to satisfy borrow checker when 'props' is mutated later.
@@ -819,60 +823,74 @@ pub async fn update_vod_metadata(
 
             // Acquire Connection logic
             let temp_handle = if active_handle.is_some() {
-                None // No new handle needed
+                None
             } else {
-                active_provider.acquire_connection_for_probe(&input.name, probe_priority).await
+                active_provider
+                    .acquire_connection_for_probe(&input.name, probe_priority)
+                    .await
+                    .map(|handle| ProbeHandleGuard::new(active_provider, handle))
             };
 
-                if active_handle.is_some() || temp_handle.is_some() {
-                    debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id);
-                    let cancel_token = select_cancel_token(temp_handle.as_ref(), active_handle);
-                    match crate::utils::ffmpeg::probe_url_with_cancel(
-                        &stream_url,
-                        user_agent.as_deref(),
-                        analyze_duration,
-                        probe_size,
-                        ffprobe_timeout,
-                        config.proxy.as_ref(),
-                        cancel_token,
-                    )
-                    .await
-                    {
-                        ProbeUrlOutcome::Success(_quality, raw_video, raw_audio) => {
-                            if let Some(details) = properties.details.as_mut() {
-                                if let Some(v) = raw_video {
-                                    details.video = Some(v.to_string().into());
-                                    properties_updated = true;
-                                }
-                                if let Some(a) = raw_audio {
-                                    details.audio = Some(a.to_string().into());
-                                    properties_updated = true;
-                                }
+            if active_handle.is_some() || temp_handle.is_some() {
+                debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id);
+                let cancel_token = select_cancel_token(
+                    temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
+                    active_handle,
+                );
+                match crate::utils::ffmpeg::probe_url_with_cancel(
+                    &stream_url,
+                    user_agent.as_deref(),
+                    analyze_duration,
+                    probe_size,
+                    ffprobe_timeout,
+                    config.proxy.as_ref(),
+                    cancel_token,
+                )
+                .await
+                {
+                    ProbeUrlOutcome::Success(_quality, raw_video, raw_audio) => {
+                        if let Some(details) = properties.details.as_mut() {
+                            if let Some(v) = raw_video {
+                                details.video = Some(v.to_string().into());
+                                properties_updated = true;
                             }
-                        }
-                        ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => {
-                            probe_failure = Some(ProbeFailureKind::NotFound);
-                        }
-                        ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => {
-                            if probe_failure.is_none() {
-                                probe_failure = Some(ProbeFailureKind::Other);
-                            }
-                        }
-                        ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => {
-                            if probe_failure.is_none() {
-                                probe_failure = Some(ProbeFailureKind::Cancelled);
+                            if let Some(a) = raw_audio {
+                                details.audio = Some(a.to_string().into());
+                                properties_updated = true;
                             }
                         }
                     }
-                if let Some(h) = temp_handle {
-                    active_provider.release_handle(&h).await;
+                    ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => {
+                        probe_failure = Some(ProbeFailureKind::NotFound);
+                    }
+                    ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => {
+                        probe_pending = true;
+                        if probe_failure.is_none() {
+                            probe_failure = Some(ProbeFailureKind::Other);
+                        }
+                    }
+                    ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => {
+                        probe_pending = true;
+                        if probe_failure.is_none() {
+                            probe_failure = Some(ProbeFailureKind::Cancelled);
+                        }
+                    }
                 }
             } else {
+                probe_pending = true;
                 warn!("Skipping probe for VOD {display_id} due to connection limits");
             }
+
+            if let Some(guard) = temp_handle {
+                guard.release().await;
+            }
         }
     }
 
+    if let Some(out) = probe_pending_out {
+        out.store(probe_pending, Ordering::Relaxed);
+    }
+
     let probe_only_unresolved = do_probe && !fetch_info && !resolve_tmdb && !properties_updated && !fetched_new;
     if probe_only_unresolved {
         if let Some(kind) = probe_failure {