Improve progress reporting and media source buffering

neckaros · neckaros · commit 2d704c5c2b8b · 2026-04-29T10:41:49.000+02:00
- Rename remaining ETA field in SSE/media convert payloads
- Add progress estimation and retry handling for media imports
- Increase stream and file I/O buffers for source providers
diff --git a/docs/SSE.md b/docs/SSE.md
@@ -122,18 +122,41 @@ interface Relations {
   booksDetails?: Book[];
 }
 
+type RsProgressType =
+  | 'download'
+  | 'transfert'
+  | 'analysing'
+  | 'finished'
+  | { duplicate: string };
+
+interface RsProgress {
+  id: string;
+  total?: number;
+  current?: number;
+  filename?: string;
+  type: RsProgressType;
+}
+
 interface UploadProgressMessage {
   library: string;
-  mediaId: string;
-  progress: number;
-  // ... additional fields
+  progress: RsProgress;
+  remainingSecondes?: number;
+}
+
+interface ConvertProgress {
+  id: string;
+  filename: string;
+  convertedId?: string | null;
+  done: boolean;
+  percent: number;
+  status: string;
+  remainingSecondes?: number | null;
+  request?: VideoConvertRequest | null;
 }
 
 interface ConvertMessage {
   library: string;
-  mediaId: string;
-  progress: number;
-  status: string;
+  progress: ConvertProgress;
 }
 
 // Content events
@@ -314,8 +337,9 @@ eventSource.addEventListener('library-status', (event) => {
 eventSource.addEventListener('convert_progress', (event) => {
   const data: SseEvent = JSON.parse(event.data);
   if ('ConvertProgress' in data) {
-    const { mediaId, progress, status } = data.ConvertProgress;
-    console.log(`Converting ${mediaId}: ${progress}% - ${status}`);
+    const { library, progress } = data.ConvertProgress;
+    const eta = progress.remainingSecondes ? `, ~${progress.remainingSecondes}s remaining` : '';
+    console.log(`Converting in ${library}: ${progress.filename} ${(progress.percent * 100).toFixed(2)}% - ${progress.status}${eta}`);
   }
 });
 
diff --git a/src/domain/media.rs b/src/domain/media.rs
@@ -172,7 +172,7 @@ pub struct ConvertProgress {
     pub done: bool,
     pub percent: f64,
     pub status: RsVideoTranscodeStatus,
-    pub estimated_remaining_seconds: Option<u64>,
+    pub remaining_secondes: Option<u64>,
     pub request: Option<VideoConvertRequest>,
 }
 
@@ -196,3 +196,26 @@ pub struct VideoMergeRequest {
     pub id: String,
     pub items: Vec<VideoMergeItem>,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn convert_progress_serializes_upload_eta_field_name() {
+        let progress = ConvertProgress {
+            id: "convert-1".to_string(),
+            filename: "video.mp4".to_string(),
+            converted_id: None,
+            done: false,
+            percent: 0.5,
+            status: RsVideoTranscodeStatus::Processing,
+            remaining_secondes: Some(42),
+            request: None,
+        };
+
+        let json = serde_json::to_string(&progress).unwrap();
+        assert!(json.contains("remainingSecondes"));
+        assert!(!json.contains("estimatedRemainingSeconds"));
+    }
+}
diff --git a/src/model/medias.rs b/src/model/medias.rs
@@ -2682,6 +2682,8 @@ impl ModelController {
             // Spawn background task
             tokio::spawn(async move {
                 let mut poll_interval = Duration::from_secs(2);
+                let start_time = std::time::Instant::now();
+                let mut remaining = None;
 
                 loop {
                     sleep(poll_interval).await;
@@ -2701,6 +2703,14 @@ impl ModelController {
 
                             let done = is_terminal;
                             let percent = current_status.progress;
+                            if percent > 1.0 {
+                                let percent_ratio = percent / 100.0;
+                                if percent_ratio > 0.01 {
+                                    let duration_from_start = start_time.elapsed().as_secs_f32();
+                                    let remaining_time = duration_from_start / percent_ratio * (1.0 - percent_ratio);
+                                    remaining = Some(remaining_time as u64);
+                                }
+                            }
 
                             let message = ConvertMessage {
                                 library: lib_progress.clone(),
@@ -2712,7 +2722,7 @@ impl ModelController {
                                     status: current_status.status.clone(),
                                     id: request_clone.id.clone(),
                                     request: Some(request_clone.clone()),
-                                    estimated_remaining_seconds: None,
+                                    remaining_secondes: remaining,
                                 },
                             };
                             mc_progress.send_convert_progress(message);
@@ -2745,12 +2755,17 @@ impl ModelController {
                                             {
                                                 Ok(reader) => {
                                                     attempts += 1;
+                                                    let mut converted_infos = media_progress.clone();
+                                                    converted_infos.size = reader.size;
+                                                    if reader.mime.is_some() {
+                                                        converted_infos.mimetype = reader.mime.clone();
+                                                    }
 
                                                     match mc_progress
                                                         .add_library_file(
                                                             &lib_progress,
                                                             &name_progress,
-                                                            Some(media_progress.clone()),
+                                                            Some(converted_infos),
                                                             reader.stream,
                                                             &ConnectedUser::ServerAdmin,
                                                         )
@@ -2773,6 +2788,7 @@ impl ModelController {
                                                     }
                                                 }
                                                 Err(e) => {
+                                                    attempts += 1;
                                                     log_error(
                                                         crate::tools::log::LogServiceType::Source,
                                                         format!("Failed to create reader: {:?}", e),
@@ -2781,6 +2797,9 @@ impl ModelController {
                                                         "Failed to create reader: {:?}",
                                                         e
                                                     );
+                                                    if attempts >= max_attempts {
+                                                        break Err(e);
+                                                    }
                                                 }
                                             }
                                         };
@@ -2981,7 +3000,7 @@ impl ModelController {
                         status: RsVideoTranscodeStatus::Processing,
                         id: request_progress.id.clone(),
                         request: Some(request_progress.clone()),
-                        estimated_remaining_seconds: remaining,
+                        remaining_secondes: remaining,
                     },
                 };
                 mc_progress.send_convert_progress(message);
@@ -3145,7 +3164,7 @@ impl ModelController {
                 done: false,
                 percent: 0.0,
                 status: RsVideoTranscodeStatus::Pending,
-                estimated_remaining_seconds: None,
+                remaining_secondes: None,
                 request: None,
             },
         };
@@ -3178,7 +3197,7 @@ impl ModelController {
                             status: RsVideoTranscodeStatus::Processing,
                             id: request_id.clone(),
                             request: None,
-                            estimated_remaining_seconds: None,
+                            remaining_secondes: None,
                         },
                     };
                     mc_progress.send_convert_progress(message);
@@ -3239,7 +3258,7 @@ impl ModelController {
                             status: RsVideoTranscodeStatus::Processing,
                             id: request_id.clone(),
                             request: None,
-                            estimated_remaining_seconds: None,
+                            remaining_secondes: None,
                         },
                     };
                     mc_progress.send_convert_progress(message);
@@ -3283,7 +3302,7 @@ impl ModelController {
                 done: false,
                 percent: 0.95,
                 status: RsVideoTranscodeStatus::Processing,
-                estimated_remaining_seconds: None,
+                remaining_secondes: None,
                 request: None,
             },
         });
@@ -3356,7 +3375,7 @@ impl ModelController {
                         done: true,
                         percent: 1.0,
                         status: RsVideoTranscodeStatus::Completed,
-                        estimated_remaining_seconds: None,
+                        remaining_secondes: None,
                         request: None,
                     },
                 });
@@ -3373,7 +3392,7 @@ impl ModelController {
                         done: true,
                         percent: 0.0,
                         status: RsVideoTranscodeStatus::Failed,
-                        estimated_remaining_seconds: None,
+                        remaining_secondes: None,
                         request: None,
                     },
                 });
diff --git a/src/model/mod.rs b/src/model/mod.rs
@@ -109,7 +109,7 @@ impl VideoConvertQueueElement {
                 done: false,
                 percent: 0f64,
                 status: RsVideoTranscodeStatus::Pending,
-                estimated_remaining_seconds: None,
+                remaining_secondes: None,
                 request: Some(request.clone()),
             },
             request,
diff --git a/src/model/plugins/video_convert_plugin.rs b/src/model/plugins/video_convert_plugin.rs
@@ -81,9 +81,28 @@ impl ModelController {
         
         let plugin_with_credentials= self.get_plugin_with_credential(plugin_id).await?;
         let temp_url =ModelController::get_temporary_read_url(library_id, media_id, Some(21600)).await?;
+        let media = self
+            .store
+            .get_library_store(library_id)?
+            .get_media(media_id, None)
+            .await?;
+        let source = if let Some(media) = media {
+            RsRequest {
+                url: temp_url,
+                size: media.item.size,
+                mime: Some(media.item.mimetype),
+                filename: Some(media.item.name),
+                ..Default::default()
+            }
+        } else {
+            RsRequest {
+                url: temp_url,
+                ..Default::default()
+            }
+        };
         let jobrequest = RsVideoTranscodeJobPluginRequest {
             job: RsVideoTranscodeJob {
-                source: RsRequest { url: temp_url, ..Default::default() },
+                source,
                 request
             },
             credentials: plugin_with_credentials.credential.clone().map(|r| r.into()).unwrap_or_default(),
@@ -224,4 +243,4 @@ impl PluginManager {
             return Err(crate::error::RsError::PluginNotFound(plugin_with_cred.plugin.id));
         }
     }
-}
+}
diff --git a/src/plugins/sources/async_reader_progress.rs b/src/plugins/sources/async_reader_progress.rs
@@ -5,9 +5,13 @@ use std::task::{Context, Poll};
 use bytes::Bytes;
 
 use crate::domain::progress::RsProgress;
+
+const PROGRESS_UPDATE_BYTES: usize = 4 * 1024 * 1024;
+
 pub struct ProgressReader<R> {
     pub inner: R,
     pub bytes_read: usize,
+    pub bytes_reported: usize,
     pub progress_template: RsProgress,
     pub sender: Sender<RsProgress>
 }
@@ -21,7 +25,7 @@ impl<R> Drop for ProgressReader<R> {
         new_progress.total = Some(self.bytes_read as u64);
 
         tokio::spawn(async move {
-            sender.send(new_progress).await.unwrap();
+            let _ = sender.send(new_progress).await;
         });
     }
 }
@@ -32,7 +36,8 @@ impl<R> ProgressReader<R> {
             inner,
             progress_template,
             sender,
-            bytes_read: 0
+            bytes_read: 0,
+            bytes_reported: 0
         }
     }
 }
@@ -43,20 +48,19 @@ impl<R: AsyncRead + Unpin> AsyncRead for ProgressReader<R> {
         cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
+        let before = buf.filled().len();
         let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
         if let Poll::Ready(Ok(())) = poll {
-            //println!("progress: {}", self.bytes_read);
-            self.bytes_read += buf.filled().len();
+            let read = buf.filled().len().saturating_sub(before);
+            self.bytes_read += read;
             let mut new_progress = self.progress_template.clone();
             new_progress.current = Some(self.bytes_read as u64); 
-            if buf.filled().len() > 0 {
-                let sender = self.sender.clone();
-                tokio::spawn(async move {
-                    sender.send(new_progress).await.unwrap();
-                });
+            if read > 0 && self.bytes_read.saturating_sub(self.bytes_reported) >= PROGRESS_UPDATE_BYTES {
+                self.bytes_reported = self.bytes_read;
+                let _ = self.sender.try_send(new_progress);
             }
             
         }
         poll
     }
-}
+}
diff --git a/src/plugins/sources/mod.rs b/src/plugins/sources/mod.rs
@@ -26,6 +26,8 @@ pub mod async_reader_progress;
 
 pub type AsyncReadPinBox = Pin<Box<dyn AsyncRead + Send + Sync>>;
 
+const STREAM_BUFFER_SIZE: usize = 4 * 1024 * 1024;
+
 pub trait AsyncSeekableWrite: AsyncWrite + AsyncSeek + Send {}
 
 impl<T> AsyncSeekableWrite for T where T: AsyncWrite + AsyncSeek + Send {}
@@ -327,7 +329,7 @@ impl SourceRead {
         match self {
             SourceRead::Stream(reader) => {
                 let headers = reader.hearders().map_err(|_| Error::UnableToFormatHeaders)?;
-                let stream = ReaderStream::new(reader.stream);
+                let stream = ReaderStream::with_capacity(reader.stream, STREAM_BUFFER_SIZE);
                 let body = Body::from_stream(stream);
                 let status = if reader.range.is_some() { axum::http::StatusCode::PARTIAL_CONTENT } else { axum::http::StatusCode::OK };
                 Ok((status, headers, body).into_response())
@@ -487,4 +489,4 @@ mod tests {
         assert_eq!(parsed.end, Some(1023), "test end parsing");
         Ok(())
     }
-}
+}
diff --git a/src/plugins/sources/path_provider.rs b/src/plugins/sources/path_provider.rs
diff --git a/src/routes/medias.rs b/src/routes/medias.rs
