mirror of
https://github.com/storytold/pdfcraft.git
synced 2026-11-10 21:11:52 +00:00
fix(open): submit password retries with priority when worker queue is full
This commit is contained in:
1 parent
acae521115
commit
6df356b63d
3 files changed
+117
-17
No files matched your search
@@ -105,6 +105,16 @@ impl PdfCraftApp {
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
pub(crate) fn schedule_open(&mut self, request: OpenRequest) -> Result<u64, String> {
|
||||
self.schedule_open_with_priority(request, false)
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
pub(crate) fn schedule_open_priority(&mut self, request: OpenRequest) -> Result<u64, String> {
|
||||
self.schedule_open_with_priority(request, true)
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn schedule_open_with_priority(&mut self, request: OpenRequest, priority: bool) -> Result<u64, String> {
|
||||
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,
|
||||
|
||||
@@ -83,29 +83,67 @@ impl OpenWorker {
|
||||
password: Option<String>,
|
||||
input: Input,
|
||||
) -> Result<Receiver<ResultMessage>, 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<String>,
|
||||
password: Option<String>,
|
||||
input: Input,
|
||||
) -> Result<Receiver<ResultMessage>, 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<String>,
|
||||
password: Option<String>,
|
||||
input: Input,
|
||||
) -> (Job, Receiver<ResultMessage>) {
|
||||
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(())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
Reference in new issue
Block a user