queue: use notify_one so wakeups are never lost

notify_waiters drops the notification when every worker is between its
DB reads and registering notified(); a task enqueued in that window sat
until a stale timer fired. notify_one stores a permit, so the next
worker to wait wakes immediately and re-leases. stop() still wakes all
workers with notify_waiters.
This commit is contained in:
2026-08-08 20:08:12 +08:00
parent aa3083792a
commit 51cc079a85
+7 -4
View File
@@ -203,9 +203,11 @@ impl PersistentTaskQueue {
Ok(()) Ok(())
}) })
.await?; .await?;
// Wake every sleeping worker: with several workers the one that finds // `notify_one` stores a permit when no worker is registered, so a
// nothing due must not starve the newly inserted row. // notification fired between a worker's DB reads and its `notified()`
self.notify.notify_waiters(); // registration is not lost (notify_waiters would drop it). The
// awakened worker re-leases and finds the new row.
self.notify.notify_one();
Ok(()) Ok(())
} }
@@ -400,7 +402,8 @@ impl QueueWorker {
if let Err(e) = result { if let Err(e) = result {
log::error!("queue reschedule failed: {e}"); log::error!("queue reschedule failed: {e}");
} }
self.notify.notify_waiters(); // Same permit semantics as enqueue: never lose the wakeup.
self.notify.notify_one();
} }
} }