fix(search): reclaim interrupted index jobs on worker start
a crash or restart left rows stuck in processing, and run_once only claims queued rows, so those repositories were never indexed again. Stale locks (over ten minutes) are requeued at startup; a freshly locked row is left alone so multiple instances cannot steal each other's work.
ZhenYi committed
Modified lib/search/src/worker.rs lib/search/src/worker.rs +41 −0 View lib/search/src/worker.rs
@@ -13,6 +13,9 @@
1313use tracing::{error, info, warn};
1414
1515use crate::outbox;
16
17/// 回收停滞 processing 任务的等待分钟数。
18const STALE_LOCK_MINUTES: i64 = 10;
1619use crate::{SearchError, SearchService};
1720
1821#[derive(Clone)]
@@ -39,6 +42,9 @@
3942 }
4043
4144 pub async fn run(self) {
45 if let Err(error) = self.requeue_interrupted().await {
46 warn!(error = %error, "search recover interrupted jobs failed");
47 }
4248 if let Err(error) = self.service.queue_initial().await {
4349 warn!(error = %error, "search initial queue failed");
4450 }
@@ -56,6 +62,41 @@
5662 info!("search worker stopped");
5763 }
5864
65 /// 进程被杀时任务会停在 processing，之后永远不会再被拾取；
66 /// 只回收锁定超过 STALE_LOCK_MINUTES 的记录，避免多实例互抢在途任务。
67 async fn requeue_interrupted(&self) -> Result<(), SearchError> {
68 let now = OffsetDateTime::now_utc();
69 let stale_before = now - time::Duration::minutes(STALE_LOCK_MINUTES);
70 let recovered = index_job::Entity::update_many()
71 .col_expr(
72 index_job::Column::State,
73 sea_orm::sea_query::Expr::value("queued"),
74 )
75 .col_expr(
76 index_job::Column::LockedAt,
77 sea_orm::sea_query::Expr::value(Option::<OffsetDateTime>::None),
78 )
79 .col_expr(
80 index_job::Column::RunAfter,
81 sea_orm::sea_query::Expr::value(now),
82 )
83 .col_expr(
84 index_job::Column::UpdatedAt,
85 sea_orm::sea_query::Expr::value(now),
86 )
87 .filter(index_job::Column::State.eq("processing"))
88 .filter(index_job::Column::LockedAt.lt(stale_before))
89 .exec(self.db.write())
90 .await?;
91 if recovered.rows_affected > 0 {
92 info!(
93 count = recovered.rows_affected,
94 "search requeued interrupted index jobs"
95 );
96 }
97 Ok(())
98 }
99
59100 pub async fn run_once(&self) -> Result<bool, SearchError> {
60101 if !self.service.enabled() {
61102 return Ok(false);