gsadmin/Firelink
1
0
Fork 0
mirror of https://github.com/nimbold/Firelink.git synced 2026-08-05 17:08:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(queue): implement concurrent deduplication and safe backend detachment

Browse Source 
- Add backendRegisteredIds and backendDispatchPromises for single dispatch enforcement

- Add detach_download_for_reconfigure to safely modify properties of active downloads

- Add applyProperties logic handling completed, failed, paused, and active queues

- Add extractValidDownloadUrls and fix multi-url paste handling

- Setup vitest and add useDownloadStore unit tests for backend registration lifecycle
This commit is contained in:
NimBold
2026-06-20 11:41:55 +03:30
parent 1660dd112d
commit 894c238bb7
17 changed files with 1058 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/download.rs
+34
View File
@@ -25,6 +25,7 @@ const WRITE_BUFFER_CAPACITY: usize = 256 * 1024;
pub enum DownloadCmd {
Start(Box<DownloadPayload>),
Pause(Uuid),
PauseWithAck(Uuid, tokio::sync::oneshot::Sender<()>),
Cancel(Uuid),
CaptureUrls(Vec<String>),
FrontendReady(bool),
@@ -116,6 +117,13 @@ impl DownloadCoordinator {
.map_err(|_| "download coordinator is unavailable".to_string())
}
pub async fn pause_media_with_ack(&self, id: String, ack: tokio::sync::oneshot::Sender<()>) -> Result<(), String> {
self.media_tx
.send(MediaCmd::PauseWithAck(id, ack))
.await
.map_err(|_| "download coordinator is unavailable".to_string())
}
pub async fn finish_media(&self, id: String) {
let _ = self.media_tx.send(MediaCmd::Finished(id)).await;
}
@@ -243,6 +251,7 @@ enum MediaCmd {
cancel_tx: watch::Sender<bool>,
},
Pause(String),
PauseWithAck(String, tokio::sync::oneshot::Sender<()>),
Finished(String),
}
@@ -281,6 +290,8 @@ async fn run_coordinator(
let (worker_tx, mut worker_rx) = mpsc::channel(128);
let mut active = HashMap::<Uuid, ActiveDownload>::new();
let mut active_media = HashMap::<String, watch::Sender<bool>>::new();
let mut pending_acks = HashMap::<Uuid, tokio::sync::oneshot::Sender<()>>::new();
let mut pending_media_acks = HashMap::<String, tokio::sync::oneshot::Sender<()>>::new();
let mut pending_captured_urls = Vec::<String>::new();
let mut frontend_ready = false;
let mut next_generation = 0_u64;
@@ -319,6 +330,14 @@ async fn run_coordinator(
let _ = download.control_tx.send(DownloadControl::Pause).await;
}
}
DownloadCmd::PauseWithAck(id, ack) => {
if let Some(download) = active.remove(&id) {
let _ = download.control_tx.send(DownloadControl::Pause).await;
pending_acks.insert(id, ack);
} else {
let _ = ack.send(());
}
}
DownloadCmd::Cancel(id) => {
if let Some(download) = active.remove(&id) {
let _ = download.control_tx.send(DownloadControl::Cancel).await;
@@ -356,6 +375,10 @@ async fn run_coordinator(
active.remove(&id);
}
if let Some(ack) = pending_acks.remove(&id) {
let _ = ack.send(());
}
match (is_current, outcome) {
(true, DownloadOutcome::Completed) => {
events.emit_completed(id);
@@ -381,8 +404,19 @@ async fn run_coordinator(
let _ = cancel_tx.send(true);
}
}
MediaCmd::PauseWithAck(id, ack) => {
if let Some(cancel_tx) = active_media.remove(&id) {
let _ = cancel_tx.send(true);
pending_media_acks.insert(id, ack);
} else {
let _ = ack.send(());
}
}
MediaCmd::Finished(id) => {
active_media.remove(&id);
if let Some(ack) = pending_media_acks.remove(&id) {
let _ = ack.send(());
}
}
}
}
src-tauri/src/extension_server.rs
+27 -2
View File
@@ -268,6 +268,7 @@ async fn enqueue_extension_download(
is_media: Some(false),
media_format_selector: None,
queue_id: MAIN_QUEUE_ID.to_string(),
has_been_dispatched: Some(true),
};
let task = crate::queue::EnqueueItem {
id,
@@ -292,12 +293,36 @@ async fn enqueue_extension_download(
is_media: Some(false),
}
.into_task();
if let Err(e) = crate::download_ownership::register_expected(
app_handle,
&item.id,
item.destination.as_deref().unwrap_or(""),
&item.file_name,
) {
log::warn!("extension: ownership registration failed: {}", e);
continue;
}
created_items.push(item);
tasks.push(task);
}
state.queue_manager.enqueue_many(tasks).await;
let _ = app_handle.emit("extension-downloads-queued", created_items);
let results = state.queue_manager.enqueue_many(tasks).await;
let mut emitted_items = Vec::new();
for (item, result) in created_items.into_iter().zip(results) {
if result.success {
emitted_items.push(item);
} else {
let _ = crate::download_ownership::remove(app_handle, &item.id);
state.queue_manager.release_registered_id(&item.id).await;
}
}
if !emitted_items.is_empty() {
let _ = app_handle.emit("extension-downloads-queued", emitted_items);
}
Ok(())
}
src-tauri/src/ipc.rs
+12
View File
@@ -95,6 +95,18 @@ pub struct DownloadItem {
#[ts(optional)]
pub media_format_selector: Option<String>,
pub queue_id: String,
#[ts(optional)]
pub has_been_dispatched: Option<bool>,
}
#[derive(Clone, Debug, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
pub struct EnqueueResult {
pub id: String,
pub success: bool,
#[ts(optional)]
pub error: Option<String>,
}
#[derive(Clone, Debug, Serialize, Deserialize, TS)]
src-tauri/src/lib.rs
+87 -5
View File
@@ -2172,6 +2172,7 @@ async fn pause_download(
state.queue_manager.clear_aria2_retry_state(&id).await;
state.queue_manager.forget_aria2_gid(&id).await;
state.queue_manager.release_permit(&id).await;
state.queue_manager.release_registered_id(&id).await;
return Err(format!(
"cannot pause aria2 gid {gid} in terminal state {terminal}"
));
@@ -2214,11 +2215,13 @@ async fn resume_download(
) -> Result<bool, String> {
let Some(gid) = state.queue_manager.aria2_gid_for_download(&id) else {
log::info!("aria2 resume [{}]: no mapped gid; re-enqueue is permitted", id);
state.queue_manager.release_registered_id(&id).await;
return Ok(false);
};
if gid.starts_with("native:") {
state.queue_manager.forget_aria2_gid(&id).await;
log::info!("aria2 resume [{}]: native fallback has no aria2 gid", id);
state.queue_manager.release_registered_id(&id).await;
return Ok(false);
}
@@ -2263,6 +2266,7 @@ async fn resume_download(
state.queue_manager.clear_aria2_retry_state(&id).await;
state.queue_manager.forget_aria2_gid(&id).await;
state.queue_manager.release_permit(&id).await;
state.queue_manager.release_registered_id(&id).await;
log::info!(
"aria2 resume [{}]: gid {} is {}; re-enqueue is permitted",
id,
@@ -2293,6 +2297,7 @@ async fn remove_download(
) -> Result<(), String> {
log::info!("remove_download called for id: {}", id);
let _ = crate::download_ownership::remove(&app_handle, &id);
state.queue_manager.release_registered_id(&id).await;
let active_kind = state.queue_manager.active_kind(&id).await;
state.queue_manager.remove_from_pending(&id).await;
@@ -2355,6 +2360,69 @@ async fn remove_download(
Ok(())
}
#[tauri::command]
async fn detach_download_for_reconfigure(
app_handle: tauri::AppHandle,
state: tauri::State<'_, AppState>,
id: String,
) -> Result<(), String> {
log::info!("detach_download_for_reconfigure called for id: {}", id);
let active_kind = state.queue_manager.active_kind(&id).await;
state.queue_manager.remove_from_pending(&id).await;
state.queue_manager.cancel_aria2_retries(&id).await;
let retry_add_guard = state.queue_manager.lock_aria2_retry_add().await;
let gid = state.queue_manager.aria2_gid_for_download(&id);
if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) {
let removal_result = async {
rpc_call(
state.aria2_port,
&state.aria2_secret,
"aria2.forcePause",
serde_json::json!([gid]),
)
.await?;
wait_for_aria2_stopped(state.aria2_port, &state.aria2_secret, gid).await
}
.await;
if let Err(error) = removal_result {
state.queue_manager.allow_aria2_retries(&id).await;
return Err(error);
}
state.queue_manager.clear_aria2_retry_state(&id).await;
state.queue_manager.forget_aria2_gid(&id).await;
state.queue_manager.release_permit(&id).await;
state.queue_manager.release_registered_id(&id).await;
log::info!("aria2 detach [{}]: gid {} stopped and forgotten", id, gid);
} else {
drop(retry_add_guard);
let (tx, rx) = tokio::sync::oneshot::channel();
if matches!(active_kind, Some(crate::queue::TaskKind::Media)) {
state.download_coordinator.pause_media_with_ack(id.clone(), tx).await?;
} else if let Ok(download_id) = Uuid::parse_str(&id) {
state.download_coordinator.send(crate::download::DownloadCmd::PauseWithAck(download_id, tx)).await?;
} else {
let _ = tx.send(()); // Fallback if no task exists
}
let _ = rx.await; // Wait for the writer to stop
if !matches!(active_kind, Some(crate::queue::TaskKind::Media)) {
state.queue_manager.release_permit(&id).await;
}
state.queue_manager.clear_aria2_retry_state(&id).await;
state.queue_manager.forget_aria2_gid(&id).await;
state.queue_manager.release_registered_id(&id).await;
}
use tauri::Emitter;
let _ = app_handle.emit(
"download-state",
crate::ipc::DownloadStateEvent::new(id.clone(), crate::ipc::DownloadStatus::Paused),
);
Ok(())
}
fn ensure_aria2_gid_result(
method: &str,
expected_gid: &str,
@@ -2506,7 +2574,11 @@ async fn enqueue_download(
&item.destination,
&item.filename,
)?;
state.queue_manager.push(item.into_task()).await;
if let Err(e) = state.queue_manager.push(item.into_task()).await {
let _ = crate::download_ownership::remove(&app_handle, &id);
state.queue_manager.release_registered_id(&id).await;
return Err(AppError::Internal(e));
}
Ok(id)
}
@@ -2515,7 +2587,7 @@ async fn enqueue_many(
app_handle: tauri::AppHandle,
state: tauri::State<'_, AppState>,
items: Vec<queue::EnqueueItem>,
) -> Result<(), AppError> {
) -> Result<Vec<crate::ipc::EnqueueResult>, AppError> {
for item in &items {
crate::download_ownership::register_expected(
&app_handle,
@@ -2525,8 +2597,16 @@ async fn enqueue_many(
)?;
}
let tasks = items.into_iter().map(queue::EnqueueItem::into_task).collect();
state.queue_manager.enqueue_many(tasks).await;
Ok(())
let results = state.queue_manager.enqueue_many(tasks).await;
for result in &results {
if !result.success {
let _ = crate::download_ownership::remove(&app_handle, &result.id);
state.queue_manager.release_registered_id(&result.id).await;
}
}
Ok(results)
}
#[tauri::command]
@@ -2546,7 +2626,8 @@ async fn remove_from_queue(
) -> Result<bool, AppError> {
let removed = state.queue_manager.remove_from_pending(&id).await;
if removed {
crate::download_ownership::remove(&app_handle, &id)?;
let _ = crate::download_ownership::remove(&app_handle, &id);
state.queue_manager.release_registered_id(&id).await;
}
Ok(removed)
}
@@ -3475,6 +3556,7 @@ pub fn run() {
set_keychain_password, get_keychain_password, delete_keychain_password,
check_file_exists, delete_file, toggle_tray_icon, set_extension_pairing_token,
set_extension_frontend_ready, set_concurrent_limit, set_global_speed_limit, remove_download,
detach_download_for_reconfigure,
enqueue_download, enqueue_many, move_in_queue, remove_from_queue, get_pending_order,
commands::reveal_in_file_manager, commands::open_downloaded_file, commands::trash_download_assets,
parity::get_system_proxy, parity::get_file_category, parity::check_for_updates, parity::is_supported_media, parity::get_supported_media_domains,
src-tauri/src/queue.rs
+45 -5
View File
@@ -86,6 +86,7 @@ pub trait SidecarSpawner: Send + Sync + 'static {
/// The centralized concurrency gatekeeper. One instance lives in AppState.
pub struct QueueManager<R: tauri::Runtime = tauri::Wry> {
registered_ids: Mutex<HashSet<String>>,
pending: Mutex<VecDeque<QueuedTask>>,
semaphore: Arc<Semaphore>,
active_permits: Mutex<HashMap<String, OwnedSemaphorePermit>>,
@@ -136,6 +137,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
spawner: Arc<dyn SidecarSpawner>,
) -> Self {
Self {
registered_ids: Mutex::new(HashSet::new()),
pending: Mutex::new(VecDeque::new()),
semaphore: Arc::new(Semaphore::new(capacity)),
active_permits: Mutex::new(HashMap::new()),
@@ -164,12 +166,25 @@ impl<R: tauri::Runtime> QueueManager<R> {
.collect()
}
/// Enqueue a task. Notifies the dispatcher. Emits download-state{queued}.
pub async fn push(&self, task: QueuedTask) {
/// Explicitly release a backend registry id (e.g. on un-resumable false paths, removals, or detach).
pub async fn release_registered_id(&self, id: &str) {
self.registered_ids.lock().await.remove(id);
}
/// Enqueue a task. Checks the centralized `registered_ids` for deduplication.
pub async fn push(&self, task: QueuedTask) -> Result<(), String> {
let id = task.id.clone();
let mut registered = self.registered_ids.lock().await;
if registered.contains(&id) {
return Err("Duplicate task".to_string());
}
registered.insert(id.clone());
drop(registered);
self.pending.lock().await.push_back(task);
self.emit_state(id, DownloadStatus::Queued);
self.notify.notify_one();
Ok(())
}
/// Pop the next task, or None if empty.
@@ -391,16 +406,20 @@ impl<R: tauri::Runtime> QueueManager<R> {
}
}
/// Called when a Media runner exits. Releases the permit and emits
/// Terminal handler for non-aria2 transfers. Emits state and frees the permit.
/// Does not emit or release anything on intentional MEDIA_RUN_CANCELLED.
/// Note: `id` is the frontend download UUID, which survives indefinitely as
/// the terminal state.
async fn finish_runner(self: Arc<Self>, id: &str, outcome: Result<(), String>) {
match outcome {
Ok(()) => {
self.emit_state(id, DownloadStatus::Completed);
self.release_registered_id(id).await;
}
Err(error) if error == MEDIA_RUN_CANCELLED => {}
Err(error) => {
self.emit_failed(id, error);
self.release_registered_id(id).await;
}
}
self.release_permit(id).await;
@@ -445,11 +464,13 @@ impl<R: tauri::Runtime> QueueManager<R> {
self.clear_aria2_retry_state(id).await;
self.forget_aria2_gid(id).await;
self.emit_state(id, DownloadStatus::Completed);
self.release_registered_id(id).await;
}
PendingOutcome::Error(error) => {
self.clear_aria2_retry_state(id).await;
self.forget_aria2_gid(id).await;
self.emit_failed(id, error);
self.release_registered_id(id).await;
}
}
self.release_permit(id).await;
@@ -736,15 +757,34 @@ impl<R: tauri::Runtime> QueueManager<R> {
}
/// Bulk enqueue by appending tasks. Used by startup and start-all.
pub async fn enqueue_many(&self, tasks: Vec<QueuedTask>) {
pub async fn enqueue_many(&self, tasks: Vec<QueuedTask>) -> Vec<crate::ipc::EnqueueResult> {
let mut results = Vec::new();
let mut registered = self.registered_ids.lock().await;
let mut pending = self.pending.lock().await;
for task in tasks {
let id = task.id.clone();
if registered.contains(&id) {
results.push(crate::ipc::EnqueueResult {
id: id.clone(),
success: false,
error: Some("Duplicate task".to_string()),
});
continue;
}
registered.insert(id.clone());
pending.push_back(task);
self.emit_state(id, DownloadStatus::Queued);
self.emit_state(id.clone(), DownloadStatus::Queued);
results.push(crate::ipc::EnqueueResult {
id,
success: true,
error: None,
});
}
drop(pending);
drop(registered);
self.notify.notify_one();
results
}
}