mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-10-01 00:32:05 +00:00
fix(queue): close the lost-wakeup window on shutdown
stop() flags the shutdown and fires notify_waiters, but a worker parked between its loop-top stop check and its notified() registration — i.e. inside earliest_run_after's DB await — was not registered when the notification fired, so it slept until the next enqueue that never comes; stop() then blocked until main's 30s shutdown timeout force-killed the drain. The sweep had the same window before recover_expired's await and would sit out a full 30s tick. Both loops now enable() the waiter first and re-check the stop flag: either the stop already happened (recheck returns) or the waiter is registered (notify_waiters reaches it) — no gap. The window itself is a scheduling race with no test seam, so this is pinned by reasoning rather than a regression test; the existing stop tests cover the ordinary path.
This commit is contained in:
@@ -179,6 +179,15 @@ impl PersistentTaskQueue {
|
|||||||
loop {
|
loop {
|
||||||
let notified = sweep_notify.notified();
|
let notified = sweep_notify.notified();
|
||||||
tokio::pin!(notified);
|
tokio::pin!(notified);
|
||||||
|
// Register before re-checking stop (same rule as the worker
|
||||||
|
// loop): a stop() landing between the previous iteration and
|
||||||
|
// here would wake nobody, and the sweep would sit out a full
|
||||||
|
// 30s tick — long enough for the shutdown timeout to treat the
|
||||||
|
// drain as stuck.
|
||||||
|
notified.as_mut().enable();
|
||||||
|
if sweep_stop.load(Ordering::Relaxed) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
_ = &mut notified => {}
|
_ = &mut notified => {}
|
||||||
_ = interval.tick() => {}
|
_ = interval.tick() => {}
|
||||||
@@ -379,6 +388,18 @@ impl QueueWorker {
|
|||||||
let wait_until = self.earliest_run_after().await;
|
let wait_until = self.earliest_run_after().await;
|
||||||
let notified = self.notify.notified();
|
let notified = self.notify.notified();
|
||||||
tokio::pin!(notified);
|
tokio::pin!(notified);
|
||||||
|
// Register before re-checking stop: a stop() between the
|
||||||
|
// loop-top check and this point fires notify_waiters with
|
||||||
|
// no waiter registered, and this worker would then sleep
|
||||||
|
// until the next enqueue (the same hazard enqueue's
|
||||||
|
// notify_one comment names). enable() closes it — either
|
||||||
|
// the stop already happened and the recheck below returns,
|
||||||
|
// or the waiter is registered and stop's notify_waiters
|
||||||
|
// reaches it.
|
||||||
|
notified.as_mut().enable();
|
||||||
|
if self.stop.load(Ordering::Relaxed) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
match wait_until {
|
match wait_until {
|
||||||
Some(until) => {
|
Some(until) => {
|
||||||
let delay = (until - now_f64()).max(0.0);
|
let delay = (until - now_f64()).max(0.0);
|
||||||
|
|||||||
Reference in New Issue
Block a user