feat(search): index repositories incrementally and add the searchhost service

Added app/searchhost/src/main.rs +48 −0 View app/searchhost/src/main.rs

@@ -0,0 +1,48 @@

1use std::path::PathBuf;

2use std::sync::Arc;

3

4use anyhow::Context;

5use config::scoped::{ScopedConfig, SynchostConfig};

6use search::{SearchService, SearchWorker};

7use tokio_util::sync::CancellationToken;

8use tracing::info;

9

10#[tokio::main]

11async fn main() -> anyhow::Result<()> {

12 export::telemetry::init_tracing();

13

14 let cfg = SynchostConfig::load().context("loading configuration")?;

15

16 info!("connecting to database");

17 let db = Arc::new(

18 export::AppDatabase::connect(cfg.database.clone())

19 .await

20 .context("connecting to postgres")?,

21 );

22

23 let search = SearchService::from_parts(db.clone(), cfg.search.clone())

24 .map_err(|e| anyhow::anyhow!("building search service: {e}"))?;

25 if !search.enabled() {

26 info!("search is disabled by config; exiting");

27 return Ok(());

28 }

29

30 let cancel = CancellationToken::new();

31 let workers = cfg.search.workers.max(1);

32 for index in 0..workers {

33 let worker = SearchWorker::new(

34 db.clone(),

35 search.clone(),

36 PathBuf::from(cfg.git.repo_root.clone()),

37 cancel.clone(),

38 );

39 tokio::spawn(worker.run());

40 info!(index, "search index worker started");

41 }

42 info!(workers, "searchhost running");

43

44 export::telemetry::shutdown_signal().await;

45 cancel.cancel();

46 info!("searchhost shut down");

47 Ok(())

48}

Added app/searchhost/Cargo.toml +14 −0 View app/searchhost/Cargo.toml

@@ -0,0 +1,14 @@

1[package]

2name = "searchhost"

3version.workspace = true

4edition.workspace = true

5publish = false

6

7[dependencies]

8config = { workspace = true }

9export = { workspace = true }

10search = { workspace = true }

11tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

12tokio-util = { workspace = true }

13tracing = { workspace = true }

14anyhow = { workspace = true }

No newline at end of file

Modified app/synchost/src/main.rs +11 −8 View app/synchost/src/main.rs

@@ -36,14 +36,17 @@

3636 let search = SearchService::from_parts(db.clone(), cfg.search.clone())

3737 .map_err(|e| anyhow::anyhow!("building search service: {e}"))?;

3838 if search.enabled() {

39 let worker = SearchWorker::new(

40 db.clone(),

41 search,

42 PathBuf::from(cfg.git.repo_root.clone()),

43 cancel.clone(),

44 );

45 tokio::spawn(worker.run());

46 info!("search index worker started");

39 let workers = cfg.search.workers.max(1);

40 for index in 0..workers {

41 let worker = SearchWorker::new(

42 db.clone(),

43 search.clone(),

44 PathBuf::from(cfg.git.repo_root.clone()),

45 cancel.clone(),

46 );

47 tokio::spawn(worker.run());

48 info!(index, "search index worker started");

49 }

4750 }

4851

4952 if cfg.import.enabled {

Modified lib/config/src/search.rs +31 −0 View lib/config/src/search.rs

@@ -7,11 +7,18 @@

77 pub url: String,

88 pub api_key: String,

99 pub index_prefix: String,

10 pub workers: usize,

1011 pub max_file_bytes: usize,

1112 pub chunk_lines: usize,

1213 pub batch_size: usize,

1314 pub poll_interval_secs: u64,

1415 pub task_timeout_secs: u64,

16 pub incremental: bool,

17 pub incremental_max_changed_files: usize,

18 pub git_timeout_secs: u64,

19 pub cat_file_batch: bool,

20 pub parse_threads: usize,

21 pub meili_inflight: usize,

1522}

1623

1724impl Default for SearchConfig {

@@ -21,11 +28,18 @@

2128 url: "http://127.0.0.1:7700".to_string(),

2229 api_key: String::new(),

2330 index_prefix: "gitdata".to_string(),

31 workers: 2,

2432 max_file_bytes: 1024 * 1024,

2533 chunk_lines: 160,

2634 batch_size: 100,

2735 poll_interval_secs: 2,

2836 task_timeout_secs: 120,

37 incremental: true,

38 incremental_max_changed_files: 2000,

39 git_timeout_secs: 120,

40 cat_file_batch: true,

41 parse_threads: 4,

42 meili_inflight: 4,

2943 }

3044 }

3145}

@@ -47,9 +61,26 @@

4761 if self.max_file_bytes == 0 || self.chunk_lines == 0 || self.batch_size == 0 {

4862 return Err("search limits must be greater than zero".to_string());

4963 }

64 if self.workers == 0 || self.workers > 64 {

65 return Err("search.workers must be between 1 and 64".to_string());

66 }

5067 if self.poll_interval_secs == 0 || self.task_timeout_secs == 0 {

5168 return Err("search intervals must be greater than zero".to_string());

5269 }

70 if self.incremental_max_changed_files == 0 {

71 return Err(

72 "search.incremental_max_changed_files must be greater than zero".to_string(),

73 );

74 }

75 if self.git_timeout_secs == 0 {

76 return Err("search.git_timeout_secs must be greater than zero".to_string());

77 }

78 if self.parse_threads == 0 || self.parse_threads > 64 {

79 return Err("search.parse_threads must be between 1 and 64".to_string());

80 }

81 if self.meili_inflight == 0 || self.meili_inflight > 64 {

82 return Err("search.meili_inflight must be between 1 and 64".to_string());

83 }

5384 Ok(())

5485 }

5586}

Modified lib/gitbaby/src/commit/helper.rs +12 −6 View lib/gitbaby/src/commit/helper.rs

@@ -322,6 +322,7 @@

322322 let mut author: Option<Signature> = None;

323323 let mut committer: Option<Signature> = None;

324324 let mut gpg_signature: Option<String> = None;

325 let mut gpg_open = false;

325326 let mut message_lines: Vec<String> = Vec::new();

326327 let mut in_message = false;

327328 for (idx, line) in text.lines().enumerate() {

@@ -336,10 +337,12 @@

336337 }

337338 if let Some(rest) = line.strip_prefix("tree ") {

338339 tree_id = Some(parse_object_id_field(rest, line_no, "tree")?);

340 gpg_open = false;

339341 continue;

340342 }

341343 if let Some(rest) = line.strip_prefix("parent ") {

342344 parents.push(parse_object_id_field(rest, line_no, "parent")?);

345 gpg_open = false;

343346 continue;

344347 }

345348 if let Some(rest) = line.strip_prefix("author ") {

@@ -349,6 +352,7 @@

349352 message: format!("bad author: {}", rest),

350353 })?,

351354 );

355 gpg_open = false;

352356 continue;

353357 }

354358 if let Some(rest) = line.strip_prefix("committer ") {

@@ -358,23 +362,25 @@

358362 message: format!("bad committer: {}", rest),

359363 })?,

360364 );

365 gpg_open = false;

361366 continue;

362367 }

363368 if let Some(rest) = line.strip_prefix("gpgsig ") {

364369 gpg_signature = Some(rest.trim_start().trim_end_matches('\r').to_string());

370 gpg_open = true;

365371 continue;

366372 }

367 if line.starts_with(' ') && gpg_signature.is_some() {

368 if let Some(sig) = gpg_signature.as_mut() {

373 if line.starts_with(' ') {

374 if gpg_open && let Some(sig) = gpg_signature.as_mut() {

369375 sig.push('\n');

370376 sig.push_str(line.trim_start());

371377 }

372378 continue;

373379 }

374 return Err(BabyError::CommitParse {

375 line_number: line_no,

376 message: format!("unexpected header line: {}", line),

377 });

380 // Unknown headers exist in the wild (`encoding`, `mergetag`, `change-id`,

381 // and other forge-specific trailers); they are irrelevant to the parsed

382 // commit info, so skip them instead of failing the whole commit.

383 gpg_open = false;

378384 }

379385 let tree_id = tree_id.ok_or_else(|| BabyError::CommitParse {

380386 line_number: 0,

Modified lib/gitbaby/src/diff/helper.rs +16 −19 View lib/gitbaby/src/diff/helper.rs

@@ -458,27 +458,24 @@

458458pub fn parse_shortstat_output(data: &[u8]) -> Result<ShortStat, BabyError> {

459459 let s = String::from_utf8_lossy(data);

460460 let mut stat = ShortStat::default();

461 let mut rest = s.trim();

462 if let Some(idx) = rest.find("files changed") {

463 stat.files_changed = parse_u64_before(rest, idx)?;

464 rest = &rest[idx + "files changed".len()..];

465 }

466 if let Some(idx) = rest.find("insertions") {

467 stat.additions = parse_u64_before(rest, idx)?;

468 rest = &rest[idx + "insertions".len()..];

469 }

470 if let Some(idx) = rest.find("deletions") {

471 stat.deletions = parse_u64_before(rest, idx)?;

461 let mut prev: Option<u64> = None;

462 for part in s.split_whitespace() {

463 if let Ok(n) = part.trim_matches(',').parse::<u64>() {

464 prev = Some(n);

465 continue;

466 }

467 let word = part

468 .trim_matches(|c: char| c == ',' || c == ')')

469 .trim_end_matches("(+")

470 .trim_end_matches("(-");

471 match word {

472 "file" | "files" => stat.files_changed = prev.take().unwrap_or(0),

473 "insertion" | "insertions" => stat.additions = prev.take().unwrap_or(0),

474 "deletion" | "deletions" => stat.deletions = prev.take().unwrap_or(0),

475 _ => {}

476 }

472477 }

473478 Ok(stat)

474}

475

476fn parse_u64_before(s: &str, before: usize) -> Result<u64, BabyError> {

477 let prefix = &s[..before];

478 let trimmed = prefix.trim().trim_end_matches(',').trim();

479 trimmed

480 .parse::<u64>()

481 .map_err(|e| BabyError::Custom(format!("invalid number `{}`: {}", trimmed, e)))

482479}

483480

484481pub fn parse_commit_diff_output(

Modified lib/models/src/search/mod.rs +1 −0 View lib/models/src/search/mod.rs

@@ -1 +1,2 @@

11pub mod index_job;

2pub mod repo_state;

Added lib/models/src/search/repo_state.rs +15 −0 View lib/models/src/search/repo_state.rs

@@ -0,0 +1,15 @@

1use sea_orm::entity::prelude::*;

2use time::OffsetDateTime;

3use uuid::Uuid;

4

5#[sea_orm::model]

6#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]

7#[sea_orm(table_name = "search_repo_state")]

8pub struct Model {

9 #[sea_orm(primary_key, auto_increment = false)]

10 pub repo_id: Uuid,

11 pub revision: String,

12 pub updated_at: OffsetDateTime,

13}

14

15impl ActiveModelBehavior for ActiveModel {}

Modified lib/models/schema_extras.sql +24 −0 View lib/models/schema_extras.sql

@@ -299,3 +299,27 @@

299299CREATE UNIQUE INDEX IF NOT EXISTS idx_ci_schedule_key ON ci_actions_schedule(repo_id,workflow_file,cron);

300300CREATE UNIQUE INDEX IF NOT EXISTS idx_ci_variable_name ON ci_variable(repo_id,name);

301301CREATE UNIQUE INDEX IF NOT EXISTS idx_ci_actions_cache_key ON ci_actions_cache(repo_id,git_ref,key,version);

302

303-- ---------------------------------------------------------------------------

304-- search index queue: coalesce duplicate queued jobs per (kind, target) and

305-- arrive-once state per repository for incremental code indexing.

306-- ---------------------------------------------------------------------------

307DELETE FROM search_index_job s

308USING search_index_job d

309WHERE s.state = 'queued'

310 AND d.state = 'queued'

311 AND s.kind = d.kind

312 AND s.target_id IS NOT DISTINCT FROM d.target_id

313 AND (s.created_at > d.created_at OR (s.created_at = d.created_at AND s.id > d.id));

314

315CREATE UNIQUE INDEX IF NOT EXISTS uniq_search_index_job_queued

316 ON search_index_job (kind, COALESCE(target_id, '00000000-0000-0000-0000-000000000000'::uuid))

317 WHERE state = 'queued';

318

319DO $$ BEGIN

320 IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'search_repo_state_repo_id_fkey') THEN

321 ALTER TABLE search_repo_state

322 ADD CONSTRAINT search_repo_state_repo_id_fkey

323 FOREIGN KEY (repo_id) REFERENCES repos(id) ON DELETE CASCADE;

324 END IF;

325END $$;

Added lib/search/src/indexer.rs +182 −0 View lib/search/src/indexer.rs

@@ -0,0 +1,182 @@

1use std::path::Path;

2use std::process::Stdio;

3use std::time::Duration;

4

5use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};

6use tokio::process::{Child, ChildStdin, ChildStdout, Command};

7

8use crate::SearchError;

9

10#[derive(Debug, Clone)]

11pub(crate) struct TreeEntry {

12 pub path: String,

13 pub size: u64,

14 pub oid: String,

15}

16

17pub(crate) async fn git_output(repo_dir: &Path, args: &[&str]) -> Result<Vec<u8>, SearchError> {

18 let output = Command::new("git")

19 .arg("--git-dir")

20 .arg(repo_dir)

21 .args(args)

22 .output()

23 .await?;

24 if output.status.success() {

25 Ok(output.stdout)

26 } else {

27 Err(SearchError::Git(

28 String::from_utf8_lossy(&output.stderr).trim().to_string(),

29 ))

30 }

31}

32

33pub(crate) async fn rev_exists(repo_dir: &Path, rev: &str) -> bool {

34 Command::new("git")

35 .arg("--git-dir")

36 .arg(repo_dir)

37 .args(["cat-file", "-e", &format!("{rev}^{{tree}}")])

38 .stdout(Stdio::null())

39 .stderr(Stdio::null())

40 .status()

41 .await

42 .map(|status| status.success())

43 .unwrap_or(false)

44}

45

46pub(crate) fn parse_ls_tree(value: &[u8]) -> Vec<TreeEntry> {

47 value

48 .split(|byte| *byte == 0)

49 .filter_map(|entry| {

50 let entry = std::str::from_utf8(entry).ok()?;

51 let (meta, path) = entry.split_once('\t')?;

52 let mut fields = meta.split_ascii_whitespace();

53 let _mode = fields.next()?;

54 let kind = fields.next()?;

55 let oid = fields.next()?;

56 let size = fields.next()?.parse().ok()?;

57 (kind == "blob").then(|| TreeEntry {

58 path: path.to_string(),

59 size,

60 oid: oid.to_string(),

61 })

62 })

63 .collect()

64}

65

66pub(crate) fn parse_name_status(value: &[u8]) -> (Vec<String>, Vec<String>) {

67 let mut changed = Vec::new();

68 let mut deleted = Vec::new();

69 let mut parts = value

70 .split(|byte| *byte == 0)

71 .filter(|part| !part.is_empty());

72 while let Some(status) = parts.next() {

73 let status = String::from_utf8_lossy(status);

74 let first = status.chars().next().unwrap_or('M');

75 if first == 'R' || first == 'C' {

76 let _source = parts.next();

77 let Some(target) = parts.next() else { break };

78 if let Ok(path) = std::str::from_utf8(target) {

79 changed.push(path.to_string());

80 }

81 } else {

82 let Some(path) = parts.next() else { break };

83 let Ok(path) = std::str::from_utf8(path) else {

84 continue;

85 };

86 if first == 'D' {

87 deleted.push(path.to_string());

88 } else {

89 changed.push(path.to_string());

90 }

91 }

92 }

93 (changed, deleted)

94}

95

96pub(crate) fn text_chunks(text: &str, chunk_lines: usize) -> Vec<(u32, u32, String)> {

97 if text.is_empty() || chunk_lines == 0 {

98 return Vec::new();

99 }

100 let lines: Vec<&str> = text.lines().collect();

101 let mut out = Vec::with_capacity(lines.len().div_ceil(chunk_lines));

102 for (index, chunk) in lines.chunks(chunk_lines).enumerate() {

103 let start = index * chunk_lines + 1;

104 let mut content = String::with_capacity(chunk.iter().map(|line| line.len() + 1).sum());

105 for (offset, line) in chunk.iter().enumerate() {

106 if offset > 0 {

107 content.push('\n');

108 }

109 content.push_str(line);

110 }

111 out.push((start as u32, (start + chunk.len() - 1) as u32, content));

112 }

113 out

114}

115

116pub(crate) struct GitCatFile {

117 _child: Child,

118 stdin: ChildStdin,

119 stdout: BufReader<ChildStdout>,

120 timeout: Duration,

121}

122

123impl GitCatFile {

124 pub(crate) fn spawn(repo_dir: &Path, timeout_secs: u64) -> Result<Self, SearchError> {

125 let mut child = Command::new("git")

126 .arg("--git-dir")

127 .arg(repo_dir)

128 .args(["cat-file", "--batch"])

129 .stdin(Stdio::piped())

130 .stdout(Stdio::piped())

131 .stderr(Stdio::null())

132 .kill_on_drop(true)

133 .spawn()?;

134 let stdin = child

135 .stdin

136 .take()

137 .ok_or_else(|| SearchError::Git("cat-file stdin unavailable".to_string()))?;

138 let stdout = BufReader::new(

139 child

140 .stdout

141 .take()

142 .ok_or_else(|| SearchError::Git("cat-file stdout unavailable".to_string()))?,

143 );

144 Ok(Self {

145 _child: child,

146 stdin,

147 stdout,

148 timeout: Duration::from_secs(timeout_secs.max(1)),

149 })

150 }

151

152 pub(crate) async fn read_blob(&mut self, oid: &str) -> Result<Option<Vec<u8>>, SearchError> {

153 let timeout = self.timeout;

154 tokio::time::timeout(timeout, async {

155 self.stdin.write_all(oid.as_bytes()).await?;

156 self.stdin.write_all(b"\n").await?;

157 self.stdin.flush().await?;

158

159 let mut header = String::new();

160 if self.stdout.read_line(&mut header).await? == 0 {

161 return Err(SearchError::Git("cat-file closed the stream".to_string()));

162 }

163 let mut fields = header.trim_end().split(' ');

164 let kind = fields.nth(1);

165 let size = fields.next().and_then(|value| value.parse::<usize>().ok());

166 let Some(size) = size else {

167 return Ok(None);

168 };

169 let mut content = vec![0u8; size];

170 self.stdout.read_exact(&mut content).await?;

171 let mut trailer = [0u8; 1];

172 self.stdout.read_exact(&mut trailer).await?;

173 if kind == Some("blob") {

174 Ok(Some(content))

175 } else {

176 Ok(None)

177 }

178 })

179 .await

180 .map_err(|_| SearchError::Git("cat-file read timed out".to_string()))?

181 }

182}

Modified lib/search/src/lib.rs +1 −0 View lib/search/src/lib.rs

@@ -1,3 +1,4 @@

1mod indexer;

12pub mod outbox;

23pub mod parser;

34pub mod service;

Modified lib/search/src/outbox.rs +19 −5 View lib/search/src/outbox.rs

@@ -1,5 +1,6 @@

1use models::search::index_job;

2use sea_orm::{ActiveModelTrait, ConnectionTrait, Set};

1use models::search::{index_job, repo_state};

2use sea_orm::sea_query::OnConflict;

3use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, Set};

34use time::OffsetDateTime;

45use uuid::Uuid;

56

@@ -10,7 +11,7 @@

1011 repo_id: Option<Uuid>,

1112) -> Result<(), sea_orm::DbErr> {

1213 let now = OffsetDateTime::now_utc();

13 index_job::ActiveModel {

14 index_job::Entity::insert(index_job::ActiveModel {

1415 id: Set(Uuid::now_v7()),

1516 kind: Set(kind.to_string()),

1617 target_id: Set(target_id),

@@ -22,8 +23,9 @@

2223 last_error: Set(None),

2324 created_at: Set(now),

2425 updated_at: Set(now),

25 }

26 .insert(db)

26 })

27 .on_conflict(OnConflict::new().do_nothing().to_owned())

28 .exec(db)

2729 .await?;

2830 Ok(())

2931}

@@ -65,3 +67,15 @@

6567pub async fn enqueue_all<C: ConnectionTrait>(db: &C) -> Result<(), sea_orm::DbErr> {

6668 enqueue(db, "all", None, None).await

6769}

70

71pub async fn clear_repo_state<C: ConnectionTrait>(

72 db: &C,

73 repo_id: Option<Uuid>,

74) -> Result<(), sea_orm::DbErr> {

75 let mut delete = repo_state::Entity::delete_many();

76 if let Some(repo_id) = repo_id {

77 delete = delete.filter(repo_state::Column::RepoId.eq(repo_id));

78 }

79 delete.exec(db).await?;

80 Ok(())

81}

Modified lib/search/src/parser.rs +27 −11 View lib/search/src/parser.rs

@@ -1,5 +1,11 @@

1use std::cell::RefCell;

2

13use tree_sitter::{Language, Parser, TreeCursor};

24

5thread_local! {

6 static PARSER: RefCell<Option<(Parser, &'static str)>> = const { RefCell::new(None) };

7}

8

39#[derive(Debug, Clone)]

410pub struct ParsedSymbol {

511 pub name: String,

@@ -153,21 +159,31 @@

153159}

154160

155161pub fn symbols_for(path: &str, text: &str) -> Vec<ParsedSymbol> {

156 let Some(language) = language_label(path).and_then(tree_sitter_language) else {

162 let Some(label) = language_label(path) else {

157163 return Vec::new();

158164 };

159 let mut parser = Parser::new();

160 if parser.set_language(&language).is_err() {

161 return Vec::new();

162 }

163 let Some(tree) = parser.parse(text, None) else {

165 let Some(language) = tree_sitter_language(label) else {

164166 return Vec::new();

165167 };

166 let bytes = text.as_bytes();

167 let mut out = Vec::new();

168 let mut cursor = tree.root_node().walk();

169 visit(&mut cursor, bytes, &mut out);

170 out

168 PARSER.with(|cell| {

169 let mut guard = cell.borrow_mut();

170 let (parser, current) = guard.get_or_insert_with(|| (Parser::new(), ""));

171 if *current != label {

172 if parser.set_language(&language).is_err() {

173 *current = "";

174 return Vec::new();

175 }

176 *current = label;

177 }

178 let Some(tree) = parser.parse(text, None) else {

179 return Vec::new();

180 };

181 let bytes = text.as_bytes();

182 let mut out = Vec::new();

183 let mut cursor = tree.root_node().walk();

184 visit(&mut cursor, bytes, &mut out);

185 out

186 })

171187}

172188

173189fn visit(cursor: &mut TreeCursor<'_>, source: &[u8], out: &mut Vec<ParsedSymbol>) {

Modified lib/search/src/service.rs +472 −135 View lib/search/src/service.rs

@@ -1,6 +1,7 @@

1use std::collections::{HashMap, HashSet};

1use std::collections::{HashMap, HashSet, VecDeque};

22use std::path::Path;

3use std::time::Duration;

3use std::sync::Arc;

4use std::time::{Duration, Instant};

45

56use config::search::SearchConfig;

67use export::AppDatabase;

@@ -12,16 +13,26 @@

1213};

1314use models::repo::{repo_collaborations, repos};

1415use models::room::{room_bans, room_members, rooms};

16use models::search::repo_state;

1517use models::space::{space_members, spaces};

1618use models::users::user;

1719use reqwest::{Client, StatusCode};

18use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};

20use sea_orm::sea_query::OnConflict;

21use sea_orm::{

22 ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, QueryOrder,

23 QuerySelect, Set, Statement, TransactionTrait,

24};

1925use serde::{Deserialize, Serialize};

2026use serde_json::{Value, json};

21use tracing::warn;

27use tokio::sync::Semaphore;

28use tokio::task::{JoinHandle, JoinSet};

29use tracing::{debug, warn};

2230use utoipa::{IntoParams, ToSchema};

2331use uuid::Uuid;

2432

33use crate::indexer::{

34 GitCatFile, git_output, parse_ls_tree, parse_name_status, rev_exists, text_chunks,

35};

2536use crate::parser::{language_label, symbols_for};

2637

2738#[derive(Debug, thiserror::Error)]

@@ -390,7 +401,7 @@

390401 self.request(reqwest::Method::PATCH, &format!("/indexes/{index}/settings"))

391402 .json(&json!({

392403 "searchableAttributes": searchable,

393 "filterableAttributes": ["repo_id", "space_id", "space_name", "repo_name", "board_id", "language", "state", "definition", "room_id"],

404 "filterableAttributes": ["repo_id", "space_id", "space_name", "repo_name", "board_id", "language", "state", "definition", "room_id", "path"],

394405 "displayedAttributes": ["*"],

395406 })),

396407 )

@@ -480,7 +491,11 @@

480491 else {

481492 return Ok(());

482493 };

483 let deadline = std::time::Instant::now() + self.timeout;

494 self.wait_uid(id).await

495 }

496

497 async fn wait_uid(&self, id: u64) -> Result<(), SearchError> {

498 let deadline = Instant::now() + self.timeout;

484499 loop {

485500 let state = self

486501 .json(self.request(reqwest::Method::GET, &format!("/tasks/{id}")))

@@ -501,11 +516,53 @@

501516 .to_string();

502517 return Err(SearchError::Task { code, message });

503518 }

504 _ if std::time::Instant::now() >= deadline => return Err(SearchError::TimedOut),

519 _ if Instant::now() >= deadline => return Err(SearchError::TimedOut),

505520 _ => tokio::time::sleep(Duration::from_millis(100)).await,

506521 }

507522 }

508523 }

524

525 async fn submit_json(

526 &self,

527 method: reqwest::Method,

528 path: &str,

529 body: &Value,

530 ) -> Result<Option<u64>, SearchError> {

531 let value = self.json(self.request(method, path).json(body)).await?;

532 Ok(value

533 .get("taskUid")

534 .or_else(|| value.get("uid"))

535 .and_then(Value::as_u64))

536 }

537

538 async fn submit_documents(

539 &self,

540 index: &str,

541 documents: &[SearchDocument],

542 ) -> Result<Option<u64>, SearchError> {

543 if documents.is_empty() {

544 return Ok(None);

545 }

546 self.submit_json(

547 reqwest::Method::POST,

548 &format!("/indexes/{index}/documents"),

549 &serde_json::to_value(documents)?,

550 )

551 .await

552 }

553

554 async fn submit_delete_filter(

555 &self,

556 index: &str,

557 filter: &str,

558 ) -> Result<Option<u64>, SearchError> {

559 self.submit_json(

560 reqwest::Method::POST,

561 &format!("/indexes/{index}/documents/delete"),

562 &json!({ "filter": filter }),

563 )

564 .await

565 }

509566}

510567

511568#[derive(Clone)]

@@ -740,29 +797,149 @@

740797 warn!(%repo_id, "search code index skipped: default branch is empty");

741798 return Ok(());

742799 }

743 let meili = self.meili()?;

744 let filter = format!("repo_id = \"{}\"", repo_id);

745 meili

746 .delete_filter(&self.index(SearchKind::Code), &filter)

747 .await?;

748 meili

749 .delete_filter(&self.index(SearchKind::Symbol), &filter)

750 .await?;

800

801 let started = Instant::now();

802 let meili = self.meili()?.clone();

803 let code_index = self.index(SearchKind::Code);

804 let symbol_index = self.index(SearchKind::Symbol);

805

806 let txn = self.db.write().begin().await?;

807 txn.query_one_raw(Statement::from_string(

808 DatabaseBackend::Postgres,

809 format!("SELECT pg_advisory_xact_lock(hashtext('search:repo:{repo_id}'))"),

810 ))

811 .await?;

812

813 let previous = repo_state::Entity::find_by_id(repo_id)

814 .one(&txn)

815 .await?

816 .map(|row| row.revision);

817

818 let mut full = !self.config.incremental;

819 let mut changed: Vec<String> = Vec::new();

820 let mut deleted: Vec<String> = Vec::new();

821 if !full {

822 match previous.as_deref() {

823 None => full = true,

824 Some(old) if old == sha => {

825 txn.commit().await?;

826 debug!(%repo_id, "search repo index already current");

827 return Ok(());

828 }

829 Some(old) => {

830 if rev_exists(&repo_dir, old).await {

831 match git_output(

832 &repo_dir,

833 &["diff", "--name-status", "-z", "--no-renames", old, &sha],

834 )

835 .await

836 {

837 Ok(bytes) => {

838 let (paths, removed) = parse_name_status(&bytes);

839 if paths.len() + removed.len()

840 > self.config.incremental_max_changed_files

841 {

842 full = true;

843 } else {

844 changed = paths;

845 deleted = removed;

846 }

847 }

848 Err(error) => {

849 warn!(%repo_id, error = %error, "search repo diff failed; falling back to full rebuild");

850 full = true;

851 }

852 }

853 } else {

854 full = true;

855 }

856 }

857 }

858 }

859

751860 let tree = git_output(&repo_dir, &["ls-tree", "-r", "-z", "--long", &sha]).await?;

752 let mut code_documents = Vec::new();

753 let mut definition_documents = Vec::new();

754 let mut reference_documents = Vec::new();

755 let mut capped = false;

756 for (path, size) in tree_entries(&tree) {

757 let Some(language) = language_label(&path) else {

861 let mut entries = parse_ls_tree(&tree);

862 if !full {

863 let wanted: HashSet<&str> = changed.iter().map(String::as_str).collect();

864 entries.retain(|entry| wanted.contains(entry.path.as_str()));

865 }

866

867 let mut task_uids: Vec<u64> = Vec::new();

868 if full {

869 if let Some(uid) = meili

870 .submit_delete_filter(&code_index, &format!("repo_id = \"{repo_id}\""))

871 .await?

872 {

873 task_uids.push(uid);

874 }

875 if let Some(uid) = meili

876 .submit_delete_filter(&symbol_index, &format!("repo_id = \"{repo_id}\""))

877 .await?

878 {

879 task_uids.push(uid);

880 }

881 } else {

882 let mut paths: Vec<String> = deleted.clone();

883 paths.extend(changed.iter().cloned());

884 for chunk in paths.chunks(DELETE_PATH_CHUNK) {

885 let filter = path_filter(repo_id, chunk);

886 if let Some(uid) = meili.submit_delete_filter(&code_index, &filter).await? {

887 task_uids.push(uid);

888 }

889 if let Some(uid) = meili.submit_delete_filter(&symbol_index, &filter).await? {

890 task_uids.push(uid);

891 }

892 }

893 }

894

895 let doc_context = Arc::new(DocContext {

896 repo_id,

897 space_id: context.space.id,

898 space_name: context.space.name.clone(),

899 repo_name: context.repo.name.clone(),

900 sha: sha.clone(),

901 chunk_lines: self.config.chunk_lines,

902 });

903 let max_file_bytes = self.config.max_file_bytes as u64;

904 let parse_threads = self.config.parse_threads.max(1);

905 let inflight_limit = parse_threads * 2;

906 let permits = Arc::new(Semaphore::new(parse_threads));

907 let mut handles: VecDeque<JoinHandle<(Vec<SearchDocument>, Vec<SearchDocument>)>> =

908 VecDeque::new();

909 let mut sink = IndexSink::new(

910 &meili,

911 &code_index,

912 &symbol_index,

913 self.config.batch_size.max(1),

914 );

915 let mut cat_file = if self.config.cat_file_batch {

916 Some(GitCatFile::spawn(&repo_dir, self.config.git_timeout_secs)?)

917 } else {

918 None

919 };

920

921 for entry in entries {

922 while handles.len() >= inflight_limit {

923 let Some(handle) = handles.pop_front() else {

924 break;

925 };

926 let (code_docs, symbol_docs) = handle.await.map_err(join_error)?;

927 sink.absorb(code_docs, symbol_docs).await?;

928 }

929 let Some(language) = language_label(&entry.path) else {

758930 continue;

759931 };

760 if size > self.config.max_file_bytes as u64 {

932 if entry.size > max_file_bytes {

761933 continue;

762934 }

763 let content = match git_output(&repo_dir, &["show", &format!("{sha}:{path}")]).await {

764 Ok(content) => content,

765 Err(_) => continue,

935 let content = match cat_file.as_mut() {

936 Some(reader) => reader.read_blob(&entry.oid).await?,

937 None => git_output(&repo_dir, &["show", &format!("{sha}:{}", entry.path)])

938 .await

939 .map(Some)?,

940 };

941 let Some(content) = content else {

942 continue;

766943 };

767944 if content.contains(&0) {

768945 continue;

@@ -770,75 +947,71 @@

770947 let Ok(text) = String::from_utf8(content) else {

771948 continue;

772949 };

773 for (start, end, chunk) in text_chunks(&text, self.config.chunk_lines) {

774 if code_documents.len() >= MAX_CODE_DOCUMENTS {

775 capped = true;

776 break;

777 }

778 code_documents.push(SearchDocument {

779 id: Uuid::now_v7().to_string(),

780 kind: SearchKind::Code.as_str().to_string(),

781 repo_id: Some(repo_id.to_string()),

782 space_id: Some(context.space.id.to_string()),

783 space_name: Some(context.space.name.clone()),

784 repo_name: Some(context.repo.name.clone()),

785 revision: Some(sha.clone()),

786 path: Some(path.clone()),

787 line_start: Some(start),

788 line_end: Some(end),

789 language: Some(language.to_string()),

790 title: path.clone(),

791 content: chunk,

792 ..Default::default()

793 });

794 }

795 for (ordinal, symbol) in symbols_for(&path, &text).into_iter().enumerate() {

796 let document = SearchDocument {

797 id: Uuid::now_v7().to_string(),

798 kind: SearchKind::Symbol.as_str().to_string(),

799 repo_id: Some(repo_id.to_string()),

800 space_id: Some(context.space.id.to_string()),

801 space_name: Some(context.space.name.clone()),

802 repo_name: Some(context.repo.name.clone()),

803 revision: Some(sha.clone()),

804 path: Some(path.clone()),

805 line_start: Some(symbol.line),

806 line_end: Some(symbol.end_line),

807 language: Some(language.to_string()),

808 symbol_id: Some(format!("{repo_id}:{sha}:{path}:{}:{ordinal}", symbol.line)),

809 symbol_name: Some(symbol.name.clone()),

810 symbol_kind: Some(symbol.kind.clone()),

811 definition: Some(symbol.is_definition),

812 title: symbol.name,

813 content: format!("{}:{}", path, symbol.line),

814 ..Default::default()

815 };

816 if symbol.is_definition {

817 definition_documents.push(document);

818 } else if reference_documents.len() < MAX_REFERENCE_DOCUMENTS {

819 reference_documents.push(document);

820 }

821 }

950 let Ok(permit) = permits.clone().acquire_owned().await else {

951 return Err(SearchError::Git("parse pool closed".to_string()));

952 };

953 let path = entry.path;

954 let parse_context = doc_context.clone();

955 handles.push_back(tokio::task::spawn_blocking(move || {

956 let _permit = permit;

957 parse_file(&parse_context, &path, language, &text)

958 }));

959 }

960 while let Some(handle) = handles.pop_front() {

961 let (code_docs, symbol_docs) = handle.await.map_err(join_error)?;

962 sink.absorb(code_docs, symbol_docs).await?;

963 }

964 sink.finish().await?;

965 task_uids.extend(sink.take_uids());

966

967 let wait_permits = Arc::new(Semaphore::new(self.config.meili_inflight.max(1)));

968 let mut waiters = JoinSet::new();

969 for uid in task_uids {

970 let Ok(permit) = wait_permits.clone().acquire_owned().await else {

971 return Err(SearchError::Git("search wait pool closed".to_string()));

972 };

973 let client = meili.clone();

974 waiters.spawn(async move {

975 let _permit = permit;

976 client.wait_uid(uid).await

977 });

978 }

979 while let Some(result) = waiters.join_next().await {

980 result.map_err(join_error)??;

822981 }

823 if capped {

982

983 repo_state::Entity::insert(repo_state::ActiveModel {

984 repo_id: Set(repo_id),

985 revision: Set(sha.clone()),

986 updated_at: Set(time::OffsetDateTime::now_utc()),

987 })

988 .on_conflict(

989 OnConflict::column(repo_state::Column::RepoId)

990 .update_columns([repo_state::Column::Revision, repo_state::Column::UpdatedAt])

991 .to_owned(),

992 )

993 .exec(&txn)

994 .await?;

995 txn.commit().await?;

996

997 if sink.capped() {

824998 warn!(

825999 %repo_id,

8261000 limit = MAX_CODE_DOCUMENTS,

8271001 "search code index reached document cap"

8281002 );

829 }

830 let mut symbol_documents = definition_documents;

831 symbol_documents.extend(reference_documents);

832 for batch in code_documents.chunks(self.config.batch_size) {

833 meili

834 .add_documents(&self.index(SearchKind::Code), batch)

835 .await?;

836 }

837 for batch in symbol_documents.chunks(self.config.batch_size) {

838 meili

839 .add_documents(&self.index(SearchKind::Symbol), batch)

840 .await?;

8411003 }

1004 debug!(

1005 %repo_id,

1006 full,

1007 files = sink.files(),

1008 changed = changed.len(),

1009 deleted = deleted.len(),

1010 code_documents = sink.code_count(),

1011 symbol_documents = sink.symbol_count(),

1012 elapsed_ms = started.elapsed().as_millis() as u64,

1013 "search repo index finished"

1014 );

8421015 Ok(())

8431016 }

8441017

@@ -1141,14 +1314,26 @@

11411314 if !self.enabled() {

11421315 return Ok(());

11431316 }

1317 // Several workers (and several processes) start at once; without a lock

1318 // they all observe an empty queue and each schedule a full rebuild, which

1319 // multiplies every index job. Serialize the check-and-enqueue with a

1320 // transaction-scoped advisory lock so exactly one caller schedules it.

1321 use sea_orm::{ConnectionTrait, DatabaseBackend, Statement, TransactionTrait};

1322 let txn = self.db.primary().begin().await?;

1323 txn.query_one_raw(Statement::from_string(

1324 DatabaseBackend::Postgres,

1325 "SELECT pg_advisory_xact_lock(721123)".to_string(),

1326 ))

1327 .await?;

11441328 let has_work = models::search::index_job::Entity::find()

11451329 .filter(models::search::index_job::Column::State.eq("queued"))

1146 .one(self.db.read())

1330 .one(&txn)

11471331 .await?

11481332 .is_some();

11491333 if !has_work {

11501334 self.queue_all().await?;

11511335 }

1336 txn.commit().await?;

11521337 Ok(())

11531338 }

11541339

@@ -1456,22 +1641,6 @@

14561641 value.replace('\\', "\\\\").replace('"', "\\\"")

14571642}

14581643

1459async fn git_output(repo_dir: &Path, args: &[&str]) -> Result<Vec<u8>, SearchError> {

1460 let output = tokio::process::Command::new("git")

1461 .arg("--git-dir")

1462 .arg(repo_dir)

1463 .args(args)

1464 .output()

1465 .await?;

1466 if output.status.success() {

1467 Ok(output.stdout)

1468 } else {

1469 Err(SearchError::Git(

1470 String::from_utf8_lossy(&output.stderr).trim().to_string(),

1471 ))

1472 }

1473}

1474

14751644/// 符号索引里“引用”条目的上限：一个文件的标识符出现次数可以非常多，不设上限时

14761645/// 大仓库会把索引撑爆并把定义挤出结果。

14771646const MAX_REFERENCE_DOCUMENTS: usize = 4_000;

@@ -1486,39 +1655,207 @@

14861655/// 回填时每次从数据库翻多少行，避免整房间一次性读进内存。

14871656const MESSAGE_BACKFILL_PAGE: u64 = 500;

14881657

1489fn tree_entries(value: &[u8]) -> Vec<(String, u64)> {

1490 value

1491 .split(|byte| *byte == 0)

1492 .filter_map(|entry| {

1493 let entry = std::str::from_utf8(entry).ok()?;

1494 let (meta, path) = entry.split_once('\t')?;

1495 let mut fields = meta.split_ascii_whitespace();

1496 let _mode = fields.next()?;

1497 let kind = fields.next()?;

1498 let _oid = fields.next()?;

1499 let size = fields.next()?.parse().ok()?;

1500 (kind == "blob").then_some((path.to_string(), size))

1501 })

1502 .collect()

1658/// 增量删除路径时每批 filter 携带的路径数，避免 Meili 请求体过大。

1659const DELETE_PATH_CHUNK: usize = 200;

1660

1661#[derive(Clone)]

1662struct DocContext {

1663 repo_id: Uuid,

1664 space_id: Uuid,

1665 space_name: String,

1666 repo_name: String,

1667 sha: String,

1668 chunk_lines: usize,

15031669}

15041670

1505fn text_chunks(text: &str, chunk_lines: usize) -> Vec<(u32, u32, String)> {

1506 let lines: Vec<&str> = text.lines().collect();

1507 if lines.is_empty() {

1508 return Vec::new();

1509 }

1510 lines

1511 .chunks(chunk_lines)

1512 .enumerate()

1513 .map(|(index, chunk)| {

1514 let start = index * chunk_lines + 1;

1515 (

1516 start as u32,

1517 (start + chunk.len() - 1) as u32,

1518 chunk.join("\n"),

1519 )

1520 })

1521 .collect()

1671fn parse_file(

1672 context: &DocContext,

1673 path: &str,

1674 language: &'static str,

1675 text: &str,

1676) -> (Vec<SearchDocument>, Vec<SearchDocument>) {

1677 let mut code_documents = Vec::new();

1678 for (start, end, chunk) in text_chunks(text, context.chunk_lines) {

1679 code_documents.push(SearchDocument {

1680 id: Uuid::now_v7().to_string(),

1681 kind: SearchKind::Code.as_str().to_string(),

1682 repo_id: Some(context.repo_id.to_string()),

1683 space_id: Some(context.space_id.to_string()),

1684 space_name: Some(context.space_name.clone()),

1685 repo_name: Some(context.repo_name.clone()),

1686 revision: Some(context.sha.clone()),

1687 path: Some(path.to_string()),

1688 line_start: Some(start),

1689 line_end: Some(end),

1690 language: Some(language.to_string()),

1691 title: path.to_string(),

1692 content: chunk,

1693 ..Default::default()

1694 });

1695 }

1696 let mut symbol_documents = Vec::new();

1697 for (ordinal, symbol) in symbols_for(path, text).into_iter().enumerate() {

1698 symbol_documents.push(SearchDocument {

1699 id: Uuid::now_v7().to_string(),

1700 kind: SearchKind::Symbol.as_str().to_string(),

1701 repo_id: Some(context.repo_id.to_string()),

1702 space_id: Some(context.space_id.to_string()),

1703 space_name: Some(context.space_name.clone()),

1704 repo_name: Some(context.repo_name.clone()),

1705 revision: Some(context.sha.clone()),

1706 path: Some(path.to_string()),

1707 line_start: Some(symbol.line),

1708 line_end: Some(symbol.end_line),

1709 language: Some(language.to_string()),

1710 symbol_id: Some(format!(

1711 "{}:{}:{}:{}:{ordinal}",

1712 context.repo_id, context.sha, path, symbol.line

1713 )),

1714 symbol_name: Some(symbol.name.clone()),

1715 symbol_kind: Some(symbol.kind.clone()),

1716 definition: Some(symbol.is_definition),

1717 title: symbol.name,

1718 content: format!("{}:{}", path, symbol.line),

1719 ..Default::default()

1720 });

1721 }

1722 (code_documents, symbol_documents)

1723}

1724

1725struct IndexSink<'a> {

1726 meili: &'a MeiliClient,

1727 code_index: &'a str,

1728 symbol_index: &'a str,

1729 batch_size: usize,

1730 pending_code: Vec<SearchDocument>,

1731 pending_symbols: Vec<SearchDocument>,

1732 uids: Vec<u64>,

1733 code_count: usize,

1734 reference_count: usize,

1735 symbol_count: usize,

1736 files: usize,

1737 capped: bool,

1738}

1739

1740impl<'a> IndexSink<'a> {

1741 fn new(

1742 meili: &'a MeiliClient,

1743 code_index: &'a str,

1744 symbol_index: &'a str,

1745 batch_size: usize,

1746 ) -> Self {

1747 Self {

1748 meili,

1749 code_index,

1750 symbol_index,

1751 batch_size,

1752 pending_code: Vec::new(),

1753 pending_symbols: Vec::new(),

1754 uids: Vec::new(),

1755 code_count: 0,

1756 reference_count: 0,

1757 symbol_count: 0,

1758 files: 0,

1759 capped: false,

1760 }

1761 }

1762

1763 async fn absorb(

1764 &mut self,

1765 code_docs: Vec<SearchDocument>,

1766 symbol_docs: Vec<SearchDocument>,

1767 ) -> Result<(), SearchError> {

1768 self.files += 1;

1769 for document in code_docs {

1770 if self.code_count >= MAX_CODE_DOCUMENTS {

1771 self.capped = true;

1772 break;

1773 }

1774 self.code_count += 1;

1775 self.pending_code.push(document);

1776 }

1777 for document in symbol_docs {

1778 if document.definition == Some(true) {

1779 self.symbol_count += 1;

1780 self.pending_symbols.push(document);

1781 } else if self.reference_count < MAX_REFERENCE_DOCUMENTS {

1782 self.reference_count += 1;

1783 self.symbol_count += 1;

1784 self.pending_symbols.push(document);

1785 }

1786 }

1787 while self.pending_code.len() >= self.batch_size {

1788 let batch: Vec<SearchDocument> = self.pending_code.drain(..self.batch_size).collect();

1789 if let Some(uid) = self.meili.submit_documents(self.code_index, &batch).await? {

1790 self.uids.push(uid);

1791 }

1792 }

1793 while self.pending_symbols.len() >= self.batch_size {

1794 let batch: Vec<SearchDocument> =

1795 self.pending_symbols.drain(..self.batch_size).collect();

1796 if let Some(uid) = self

1797 .meili

1798 .submit_documents(self.symbol_index, &batch)

1799 .await?

1800 {

1801 self.uids.push(uid);

1802 }

1803 }

1804 Ok(())

1805 }

1806

1807 async fn finish(&mut self) -> Result<(), SearchError> {

1808 if !self.pending_code.is_empty() {

1809 let batch = std::mem::take(&mut self.pending_code);

1810 if let Some(uid) = self.meili.submit_documents(self.code_index, &batch).await? {

1811 self.uids.push(uid);

1812 }

1813 }

1814 if !self.pending_symbols.is_empty() {

1815 let batch = std::mem::take(&mut self.pending_symbols);

1816 if let Some(uid) = self

1817 .meili

1818 .submit_documents(self.symbol_index, &batch)

1819 .await?

1820 {

1821 self.uids.push(uid);

1822 }

1823 }

1824 Ok(())

1825 }

1826

1827 fn take_uids(&mut self) -> Vec<u64> {

1828 std::mem::take(&mut self.uids)

1829 }

1830

1831 fn files(&self) -> usize {

1832 self.files

1833 }

1834

1835 fn code_count(&self) -> usize {

1836 self.code_count

1837 }

1838

1839 fn symbol_count(&self) -> usize {

1840 self.symbol_count

1841 }

1842

1843 fn capped(&self) -> bool {

1844 self.capped

1845 }

1846}

1847

1848fn path_filter(repo_id: Uuid, paths: &[String]) -> String {

1849 let list = paths

1850 .iter()

1851 .map(|path| format!("\"{}\"", escape_filter(path)))

1852 .collect::<Vec<_>>()

1853 .join(", ");

1854 format!("repo_id = \"{repo_id}\" AND path IN [{list}]")

1855}

1856

1857fn join_error(error: tokio::task::JoinError) -> SearchError {

1858 SearchError::Git(format!("search task join failed: {error}"))

15221859}

15231860

15241861fn compact_snippet(value: &str) -> String {

Modified lib/search/src/worker.rs +59 −7 View lib/search/src/worker.rs

@@ -8,7 +8,9 @@

88use models::repo::repos;

99use models::room::rooms;

1010use models::search::index_job;

11use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder};

11use sea_orm::{

12 ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, QueryOrder, Statement,

13};

1214use time::OffsetDateTime;

1315use tokio_util::sync::CancellationToken;

1416use tracing::{error, info, warn};

@@ -49,14 +51,28 @@

4951 if let Err(error) = self.service.queue_initial().await {

5052 warn!(error = %error, "search initial queue failed");

5153 }

52 let delay = Duration::from_secs(self.service.config().poll_interval_secs);

54 let delay = Duration::from_secs(self.service.config().poll_interval_secs.max(1));

55 let mut last_sweep = tokio::time::Instant::now();

5356 loop {

5457 tokio::select! {

5558 _ = self.cancel.cancelled() => break,

56 _ = tokio::time::sleep(delay) => match self.run_once().await {

57 Ok(true) => {}

58 Ok(false) => {}

59 Err(error) => warn!(error = %error, "search worker iteration failed"),

59 _ = tokio::time::sleep(delay) => {

60 if last_sweep.elapsed() >= Duration::from_secs(60) {

61 if let Err(error) = self.requeue_interrupted().await {

62 warn!(error = %error, "search stale job sweep failed");

63 }

64 last_sweep = tokio::time::Instant::now();

65 }

66 while !self.cancel.is_cancelled() {

67 match self.run_once().await {

68 Ok(true) => {}

69 Ok(false) => break,

70 Err(error) => {

71 warn!(error = %error, "search worker iteration failed");

72 break;

73 }

74 }

75 }

6076 }

6177 }

6278 }

@@ -68,6 +84,21 @@

6884 async fn requeue_interrupted(&self) -> Result<(), SearchError> {

6985 let now = OffsetDateTime::now_utc();

7086 let stale_before = now - time::Duration::minutes(STALE_LOCK_MINUTES);

87 // queued 去重索引下，回收时若同目标已有 queued 行，直接丢弃僵尸 processing 行，

88 // 由 queued 行按最新状态重建，避免 UPDATE 撞唯一索引。

89 self.db

90 .write()

91 .execute_raw(Statement::from_string(

92 DatabaseBackend::Postgres,

93 format!(

94 "DELETE FROM search_index_job p USING search_index_job q \

95 WHERE p.state = 'processing' AND p.locked_at < to_timestamp({}) \

96 AND q.state = 'queued' AND q.kind = p.kind \

97 AND q.target_id IS NOT DISTINCT FROM p.target_id",

98 stale_before.unix_timestamp()

99 ),

100 ))

101 .await?;

71102 let recovered = index_job::Entity::update_many()

72103 .col_expr(

73104 index_job::Column::State,

@@ -132,6 +163,7 @@

132163 if claimed.rows_affected == 0 {

133164 return Ok(false);

134165 }

166 let started = tokio::time::Instant::now();

135167 let result = self.process(&job).await;

136168 match result {

137169 Ok(()) => {

@@ -147,12 +179,32 @@

147179 .filter(index_job::Column::Id.eq(job.id))

148180 .exec(self.db.write())

149181 .await?;

150 info!(job_id = %job.id, kind = %job.kind, "search index job completed");

182 info!(

183 job_id = %job.id,

184 kind = %job.kind,

185 elapsed_ms = started.elapsed().as_millis() as u64,

186 lag_ms = (OffsetDateTime::now_utc() - job.created_at).whole_milliseconds().max(0) as u64,

187 "search index job completed"

188 );

151189 }

152190 Err(error) => {

153191 let attempts = job.attempts + 1;

154192 let retry_after = OffsetDateTime::now_utc()

155193 + time::Duration::seconds((attempts.min(8) * 5) as i64);

194 // 处理期间若同目标已有 queued 孪生行，直接丢弃本次失败行，

195 // 由孪生行按最新状态重试，避免回队时撞 queued 去重索引。

196 self.db

197 .write()

198 .execute_raw(Statement::from_string(

199 DatabaseBackend::Postgres,

200 format!(

201 "DELETE FROM search_index_job f USING search_index_job q \

202 WHERE f.id = '{}' AND q.state = 'queued' AND q.kind = f.kind \

203 AND q.target_id IS NOT DISTINCT FROM f.target_id",

204 job.id

205 ),

206 ))

207 .await?;

156208 index_job::Entity::update_many()

157209 .col_expr(

158210 index_job::Column::State,