From 6df356b63dd2411199b7506252283b04f8ea99cf Mon Sep 17 00:00:00 2001 From: Combiler01 Date: Tue, 10 Nov 2026 11:06:56 +0530 Subject: [PATCH] fix(open): submit password retries with priority when worker queue is full --- crates/ui-egui/src/app/files.rs | 19 +++++- crates/ui-egui/src/platform/open_worker.rs | 66 +++++++++++++++---- .../ui-egui/src/platform/open_worker/tests.rs | 49 ++++++++++++++ 3 files changed, 117 insertions(+), 17 deletions(-) diff --git a/crates/ui-egui/src/app/files.rs b/crates/ui-egui/src/app/files.rs index a86c72ef7d..ec264b07e3 100644 --- a/crates/ui-egui/src/app/files.rs +++ b/crates/ui-egui/src/app/files.rs @@ -105,6 +105,16 @@ impl PdfCraftApp { #[cfg(not(target_arch = "wasm32"))] pub(crate) fn schedule_open(&mut self, request: OpenRequest) -> Result { + self.schedule_open_with_priority(request, false) + } + + #[cfg(not(target_arch = "wasm32"))] + pub(crate) fn schedule_open_priority(&mut self, request: OpenRequest) -> Result { + self.schedule_open_with_priority(request, true) + } + + #[cfg(not(target_arch = "wasm32"))] + fn schedule_open_with_priority(&mut self, request: OpenRequest, priority: bool) -> Result { if self.open_worker.is_none() { self.open_worker = Some( open_worker::OpenWorker::new( @@ -119,8 +129,11 @@ impl PdfCraftApp { if let Some(ctx) = &self.ctx { worker.set_context(ctx); } - let receiver = - worker.submit(token, self.session.open_context(), request.name.clone(), request.path.clone(), request.password, request.input)?; + let receiver = if priority { + worker.submit_priority(token, self.session.open_context(), request.name.clone(), request.path.clone(), request.password, request.input)? + } else { + worker.submit(token, self.session.open_context(), request.name.clone(), request.path.clone(), request.password, request.input)? + }; self.next_open_token = token; self.pending_opens.push(PendingOpen { token, @@ -497,7 +510,7 @@ impl PdfCraftApp { }; #[cfg(not(target_arch = "wasm32"))] if self.ctx.is_some() && !self.run_inline { - let result = self.schedule_open(OpenRequest { + let result = self.schedule_open_priority(OpenRequest { name: p.name.clone(), path: p.path.clone(), stamp: p.stamp, diff --git a/crates/ui-egui/src/platform/open_worker.rs b/crates/ui-egui/src/platform/open_worker.rs index 274071ea78..24211ffc3a 100644 --- a/crates/ui-egui/src/platform/open_worker.rs +++ b/crates/ui-egui/src/platform/open_worker.rs @@ -83,29 +83,67 @@ impl OpenWorker { password: Option, input: Input, ) -> Result, String> { + let (job, receiver) = self.build_job(token, context, name, path, password, input); + self.enqueue(job, false)?; + Ok(receiver) + } + + pub fn submit_priority( + &self, + token: u64, + context: OpenContext, + name: String, + path: Option, + password: Option, + input: Input, + ) -> Result, String> { + let (job, receiver) = self.build_job(token, context, name, path, password, input); + self.enqueue(job, true)?; + Ok(receiver) + } + + fn build_job( + &self, + token: u64, + context: OpenContext, + name: String, + path: Option, + password: Option, + input: Input, + ) -> (Job, Receiver) { let (reply, receiver) = mpsc::channel(); - let job = Job { - token, - context, - name, - path, - password, - input, - reply, - #[cfg(test)] - gate: self.gate.clone(), - }; + ( + Job { + token, + context, + name, + path, + password, + input, + reply, + #[cfg(test)] + gate: self.gate.clone(), + }, + receiver, + ) + } + + fn enqueue(&self, job: Job, priority: bool) -> Result<(), String> { let (lock, ready) = &*self.queue; let mut queue = lock.lock().unwrap_or_else(PoisonError::into_inner); if queue.shutdown { return Err("the document-open worker has stopped".into()); } - if queue.jobs.len() >= MAX_QUEUED_OPENS { + if !priority && queue.jobs.len() >= MAX_QUEUED_OPENS { return Err("too many documents are waiting to open".into()); } - queue.jobs.push_back(job); + if priority { + queue.jobs.push_front(job); + } else { + queue.jobs.push_back(job); + } ready.notify_one(); - Ok(receiver) + Ok(()) } } diff --git a/crates/ui-egui/src/platform/open_worker/tests.rs b/crates/ui-egui/src/platform/open_worker/tests.rs index efe911bff6..a39bb9fd4e 100644 --- a/crates/ui-egui/src/platform/open_worker/tests.rs +++ b/crates/ui-egui/src/platform/open_worker/tests.rs @@ -921,3 +921,52 @@ fn a_repeated_waiting_open_keeps_the_tool_it_brings() { assert_eq!(app.open_path(&path), crate::OpenOutcome::Pending); assert_eq!(app.deferred_opens.front().and_then(|d| d.then), Some("export.text")); } + +/// A password retry submits with priority and succeeds even when the open worker queue is full +/// of ordinary background documents, placing the retry at the head of the worker queue. +#[test] +#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +fn password_retry_submits_with_priority_when_worker_queue_is_full() { + let (gate, started, release, finished) = open_worker::prepare_gate(); + let worker = open_worker::OpenWorker::new(Some(gate)).expect("worker"); + let pdf = Arc::new(fixture(1)); + let open_context = pdfcraft_engine::Session::new().open_context(); + + // Fill the worker queue to its limit while the gate holds the worker on the first job. + let _first = + worker.submit(1, open_context.clone(), "first.pdf".into(), None, None, open_worker::Input::Bytes(pdf.clone())).expect("submit first"); + started.recv_timeout(Duration::from_secs(5)).expect("first job reached gate"); + + for token in 2..=(open_worker::MAX_QUEUED_OPENS as u64 + 1) { + worker + .submit(token, open_context.clone(), format!("doc-{token}.pdf"), None, None, open_worker::Input::Bytes(pdf.clone())) + .expect("queue up to limit"); + } + + // At capacity, an ordinary submit fails with the capacity error. + let overflow = worker.submit(999, open_context.clone(), "overflow.pdf".into(), None, None, open_worker::Input::Bytes(pdf.clone())); + assert_eq!(overflow.err(), Some("too many documents are waiting to open".into())); + + // A priority submit (such as a password prompt retry) succeeds and enters the queue. + let priority_token = 1000; + let priority_recv = worker.submit_priority( + priority_token, + open_context.clone(), + "password-retry.pdf".into(), + None, + Some("secret".into()), + open_worker::Input::Bytes(pdf.clone()), + ); + assert!(priority_recv.is_ok(), "priority submit must bypass the queued opens limit"); + + // Release the first job; the priority job must be processed next because it was pushed to the front. + release.send(()).expect("release first"); + finished.recv_timeout(Duration::from_secs(5)).expect("first finished"); + + started.recv_timeout(Duration::from_secs(5)).expect("priority job reached gate next"); + release.send(()).expect("release priority"); + finished.recv_timeout(Duration::from_secs(5)).expect("priority finished"); + + let priority_result = priority_recv.unwrap().recv_timeout(Duration::from_secs(5)).expect("priority result"); + assert_eq!(priority_result.token, priority_token); +}