From 8f2b7ee26750c1afb4bfa0ee1a4b28515427fdf5 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Fri, 25 Sep 2026 16:54:35 +0900 Subject: [PATCH] gitmirror: language index Signed-off-by: Seongmin Lee --- Cargo.lock | 23 ++ Cargo.toml | 1 + ...8d5665152b7431776b565fa9ed9324090a78d.json | 16 ++ ...20a0367e03cd610a742b878b416eeb9a3f561.json | 65 ++++++ ...9e61a7c74c77ae3418ea43a9bc929d38f324e.json | 14 ++ ...a4711e0805a771b398f78d8cc43fe15f2758.json} | 5 +- ...0891856550ef9da693a41d18cd62189aad307.json | 15 ++ ...ab230c726baf267f68f625aa1cdbcba5bd312.json | 17 ++ .../migrations/20260926000000_index.sql | 18 ++ gitmirror/crates/gitmirror-db/src/lib.rs | 120 +++++++++- gitmirror/crates/gitmirror-index/Cargo.toml | 27 +++ gitmirror/crates/gitmirror-index/src/lib.rs | 209 ++++++++++++++++++ gitmirror/crates/gitmirror-sync/src/lib.rs | 33 ++- gitmirror/crates/gitmirror-sync/src/task.rs | 11 +- gitmirror/crates/gitmirror-types/Cargo.toml | 1 + gitmirror/crates/gitmirror-types/src/lib.rs | 17 ++ gitmirror/crates/gitmirror/Cargo.toml | 1 + gitmirror/crates/gitmirror/src/config.rs | 21 ++ gitmirror/crates/gitmirror/src/main.rs | 25 ++- 19 files changed, 611 insertions(+), 28 deletions(-) create mode 100644 gitmirror/crates/gitmirror-db/.sqlx/query-63518fc08ee86cdaa6a881183578d5665152b7431776b565fa9ed9324090a78d.json create mode 100644 gitmirror/crates/gitmirror-db/.sqlx/query-66df0c0bb00794d4fabfc89d3ca20a0367e03cd610a742b878b416eeb9a3f561.json create mode 100644 gitmirror/crates/gitmirror-db/.sqlx/query-7ff739bd0a1b556db369e2948df9e61a7c74c77ae3418ea43a9bc929d38f324e.json rename gitmirror/crates/gitmirror-db/.sqlx/{query-a11215321a6b36170dfaaa245329a37363881e690b4100d586b52787bf63c1ec.json => query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json} (57%) create mode 100644 gitmirror/crates/gitmirror-db/.sqlx/query-ee3e927e80d0f3051ed3d2de99c0891856550ef9da693a41d18cd62189aad307.json create mode 100644 gitmirror/crates/gitmirror-db/.sqlx/query-ef69831ad84fc5dc6989cd661cfab230c726baf267f68f625aa1cdbcba5bd312.json create mode 100644 gitmirror/crates/gitmirror-db/migrations/20260926000000_index.sql create mode 100644 gitmirror/crates/gitmirror-index/Cargo.toml create mode 100644 gitmirror/crates/gitmirror-index/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 090526033..c30951583 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2869,6 +2869,7 @@ dependencies = [ "futures", "gitmirror-db", "gitmirror-git", + "gitmirror-index", "gitmirror-ingest", "gitmirror-sync", "gitmirror-types", @@ -2914,6 +2915,27 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "gitmirror-index" +version = "0.0.1" +dependencies = [ + "bobbin-runtime", + "gitmirror-db", + "gitmirror-git", + "gitmirror-types", + "jacquard-common", + "knot-git", + "knot-langs", + "knot-types", + "reqwest 0.13.5", + "scc", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "gitmirror-ingest" version = "0.0.1" @@ -2965,6 +2987,7 @@ name = "gitmirror-types" version = "0.0.1" dependencies = [ "bobbin-knot-proxy", + "bobbin-runtime", "jacquard-common", ] diff --git a/Cargo.toml b/Cargo.toml index 04a3108ab..361c3d1be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -83,6 +83,7 @@ knot-edge = { path = "knot2/crates/knot-edge" } gitmirror-db = { path = "gitmirror/crates/gitmirror-db" } gitmirror-git = { path = "gitmirror/crates/gitmirror-git" } +gitmirror-index = { path = "gitmirror/crates/gitmirror-index" } gitmirror-ingest = { path = "gitmirror/crates/gitmirror-ingest" } gitmirror-sync = { path = "gitmirror/crates/gitmirror-sync" } gitmirror-types = { path = "gitmirror/crates/gitmirror-types" } diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-63518fc08ee86cdaa6a881183578d5665152b7431776b565fa9ed9324090a78d.json b/gitmirror/crates/gitmirror-db/.sqlx/query-63518fc08ee86cdaa6a881183578d5665152b7431776b565fa9ed9324090a78d.json new file mode 100644 index 000000000..4bb75e2e8 --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-63518fc08ee86cdaa6a881183578d5665152b7431776b565fa9ed9324090a78d.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "insert into repo_languages (did, language, size)\n select $1, * from unnest($2::text[], $3::bigint[])", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "TextArray", + "Int8Array" + ] + }, + "nullable": [] + }, + "hash": "63518fc08ee86cdaa6a881183578d5665152b7431776b565fa9ed9324090a78d" +} diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-66df0c0bb00794d4fabfc89d3ca20a0367e03cd610a742b878b416eeb9a3f561.json b/gitmirror/crates/gitmirror-db/.sqlx/query-66df0c0bb00794d4fabfc89d3ca20a0367e03cd610a742b878b416eeb9a3f561.json new file mode 100644 index 000000000..1eba488e1 --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-66df0c0bb00794d4fabfc89d3ca20a0367e03cd610a742b878b416eeb9a3f561.json @@ -0,0 +1,65 @@ +{ + "db_name": "PostgreSQL", + "query": "select did, default_branch, default_branch_commit, index_retry_count\n from repos\n where deleted_at is null\n and state = 'active'\n and default_branch_commit is distinct from indexed_commit\n and (index_retry_after_us is null or index_retry_after_us <= $1)\n order by index_retry_after_us nulls first, updated_at\n limit $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "did", + "type_info": "Text", + "origin": { + "Table": { + "table": "repos", + "name": "did" + } + } + }, + { + "ordinal": 1, + "name": "default_branch", + "type_info": "Text", + "origin": { + "Table": { + "table": "repos", + "name": "default_branch" + } + } + }, + { + "ordinal": 2, + "name": "default_branch_commit", + "type_info": "Text", + "origin": { + "Table": { + "table": "repos", + "name": "default_branch_commit" + } + } + }, + { + "ordinal": 3, + "name": "index_retry_count", + "type_info": "Int4", + "origin": { + "Table": { + "table": "repos", + "name": "index_retry_count" + } + } + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [ + false, + false, + true, + false + ] + }, + "hash": "66df0c0bb00794d4fabfc89d3ca20a0367e03cd610a742b878b416eeb9a3f561" +} diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-7ff739bd0a1b556db369e2948df9e61a7c74c77ae3418ea43a9bc929d38f324e.json b/gitmirror/crates/gitmirror-db/.sqlx/query-7ff739bd0a1b556db369e2948df9e61a7c74c77ae3418ea43a9bc929d38f324e.json new file mode 100644 index 000000000..35c8da8b3 --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-7ff739bd0a1b556db369e2948df9e61a7c74c77ae3418ea43a9bc929d38f324e.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "delete from repo_languages where did = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "7ff739bd0a1b556db369e2948df9e61a7c74c77ae3418ea43a9bc929d38f324e" +} diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-a11215321a6b36170dfaaa245329a37363881e690b4100d586b52787bf63c1ec.json b/gitmirror/crates/gitmirror-db/.sqlx/query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json similarity index 57% rename from gitmirror/crates/gitmirror-db/.sqlx/query-a11215321a6b36170dfaaa245329a37363881e690b4100d586b52787bf63c1ec.json rename to gitmirror/crates/gitmirror-db/.sqlx/query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json index 4150576dd..a71653dbe 100644 --- a/gitmirror/crates/gitmirror-db/.sqlx/query-a11215321a6b36170dfaaa245329a37363881e690b4100d586b52787bf63c1ec.json +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json @@ -1,15 +1,16 @@ { "db_name": "PostgreSQL", - "query": "update repos\n set state = 'active', error_msg = null, retry_count = 0, retry_after_us = null,\n updated_at = now()\n where did = $1 and refs_cid = $2 and deleted_at is null", + "query": "update repos\n set state = 'active', error_msg = null, retry_count = 0, retry_after_us = null,\n default_branch_commit = $3, updated_at = now()\n where did = $1 and refs_cid = $2 and deleted_at is null", "describe": { "columns": [], "parameters": { "Left": [ + "Text", "Text", "Text" ] }, "nullable": [] }, - "hash": "a11215321a6b36170dfaaa245329a37363881e690b4100d586b52787bf63c1ec" + "hash": "e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758" } diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-ee3e927e80d0f3051ed3d2de99c0891856550ef9da693a41d18cd62189aad307.json b/gitmirror/crates/gitmirror-db/.sqlx/query-ee3e927e80d0f3051ed3d2de99c0891856550ef9da693a41d18cd62189aad307.json new file mode 100644 index 000000000..8a12b47e2 --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-ee3e927e80d0f3051ed3d2de99c0891856550ef9da693a41d18cd62189aad307.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "update repos\n set indexed_commit = $2, index_error = null, index_retry_count = 0,\n index_retry_after_us = null\n where did = $1 and default_branch_commit is not distinct from $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "ee3e927e80d0f3051ed3d2de99c0891856550ef9da693a41d18cd62189aad307" +} diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-ef69831ad84fc5dc6989cd661cfab230c726baf267f68f625aa1cdbcba5bd312.json b/gitmirror/crates/gitmirror-db/.sqlx/query-ef69831ad84fc5dc6989cd661cfab230c726baf267f68f625aa1cdbcba5bd312.json new file mode 100644 index 000000000..b5803eae0 --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-ef69831ad84fc5dc6989cd661cfab230c726baf267f68f625aa1cdbcba5bd312.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "update repos\n set index_error = $3, index_retry_count = index_retry_count + 1,\n index_retry_after_us = $4\n where did = $1 and default_branch_commit is not distinct from $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "ef69831ad84fc5dc6989cd661cfab230c726baf267f68f625aa1cdbcba5bd312" +} diff --git a/gitmirror/crates/gitmirror-db/migrations/20260926000000_index.sql b/gitmirror/crates/gitmirror-db/migrations/20260926000000_index.sql new file mode 100644 index 000000000..bcbaccedf --- /dev/null +++ b/gitmirror/crates/gitmirror-db/migrations/20260926000000_index.sql @@ -0,0 +1,18 @@ +alter table repos + add column default_branch_commit text, + add column indexed_commit text, + add column index_error text, + add column index_retry_count integer not null default 0, + add column index_retry_after_us bigint; + +create index if not exists idx_repos_index_queue + on repos (index_retry_after_us nulls first, updated_at) + where deleted_at is null and state = 'active' + and default_branch_commit is distinct from indexed_commit; + +create table if not exists repo_languages ( + did text not null references repos (did), + language text not null, + size bigint not null, + primary key (did, language) +); diff --git a/gitmirror/crates/gitmirror-db/src/lib.rs b/gitmirror/crates/gitmirror-db/src/lib.rs index 6153c2add..144243328 100644 --- a/gitmirror/crates/gitmirror-db/src/lib.rs +++ b/gitmirror/crates/gitmirror-db/src/lib.rs @@ -67,6 +67,14 @@ pub struct RefsUpsert<'a> { pub refs: &'a GitRefs, } +#[derive(Debug)] +pub struct IndexRow { + pub did: Did, + pub default_branch: SmolStr, + pub default_branch_commit: Option, + pub index_retry_count: i32, +} + #[derive(Debug)] pub struct RepoRow { pub did: Did, @@ -216,14 +224,20 @@ impl Db { Ok(rows.into_iter().map(|r| (r.state, r.due, r.count)).collect()) } - pub async fn mark_synced(&self, did: &Did, refs_cid: &str) -> Result<(), DbError> { + pub async fn mark_synced( + &self, + did: &Did, + refs_cid: &str, + default_branch_commit: Option<&str>, + ) -> Result<(), DbError> { let done = sqlx::query!( "update repos set state = 'active', error_msg = null, retry_count = 0, retry_after_us = null, - updated_at = now() + default_branch_commit = $3, updated_at = now() where did = $1 and refs_cid = $2 and deleted_at is null", did.as_str(), refs_cid, + default_branch_commit, ) .execute(&self.0) .await?; @@ -290,4 +304,106 @@ impl Db { } Ok(()) } + + /// Repos whose default branch commit moved since the last index, longest-waiting first. + pub async fn pending_index(&self, now_us: i64, limit: i64) -> Result, DbError> { + let rows = sqlx::query!( + "select did, default_branch, default_branch_commit, index_retry_count + from repos + where deleted_at is null + and state = 'active' + and default_branch_commit is distinct from indexed_commit + and (index_retry_after_us is null or index_retry_after_us <= $1) + order by index_retry_after_us nulls first, updated_at + limit $2", + now_us, + limit, + ) + .fetch_all(&self.0) + .await?; + Ok(rows + .into_iter() + .filter_map(|r| { + let did = Did::new_owned(&r.did) + .inspect_err(|e| warn!(did = %r.did, error = %e, "unparseable did in repos")) + .ok()?; + Some(IndexRow { + did, + default_branch: r.default_branch.into(), + default_branch_commit: r.default_branch_commit, + index_retry_count: r.index_retry_count, + }) + }) + .collect()) + } + + pub async fn set_languages( + &self, + did: &Did, + commit: Option<&str>, + langs: &[(String, i64)], + ) -> Result<(), DbError> { + let mut tx = self.0.begin().await?; + let done = sqlx::query!( + "update repos + set indexed_commit = $2, index_error = null, index_retry_count = 0, + index_retry_after_us = null + where did = $1 and default_branch_commit is not distinct from $2", + did.as_str(), + commit, + ) + .execute(&mut *tx) + .await?; + if done.rows_affected() == 0 { + debug!(%did, ?commit, "default branch moved mid-index, dropped the stale languages"); + return Ok(()); + } + sqlx::query!("delete from repo_languages where did = $1", did.as_str()) + .execute(&mut *tx) + .await?; + let (names, sizes): (Vec<&str>, Vec) = + langs.iter().map(|(n, s)| (n.as_str(), *s)).unzip(); + sqlx::query!( + "insert into repo_languages (did, language, size) + select $1, * from unnest($2::text[], $3::bigint[])", + did.as_str(), + &names as &[&str], + &sizes, + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + debug!(%did, ?commit, langs = langs.len(), "indexed languages"); + Ok(()) + } + + pub async fn mark_index_failed( + &self, + did: &Did, + commit: Option<&str>, + error: &str, + retry_after_us: i64, + ) -> Result<(), DbError> { + let error = if error.contains('\0') { + Cow::Owned(error.replace('\0', "")) + } else { + Cow::Borrowed(error) + }; + let done = sqlx::query!( + "update repos + set index_error = $3, index_retry_count = index_retry_count + 1, + index_retry_after_us = $4 + where did = $1 and default_branch_commit is not distinct from $2", + did.as_str(), + commit, + &*error, + retry_after_us, + ) + .execute(&self.0) + .await?; + if done.rows_affected() > 0 { + warn!(%did, ?commit, retry_after_us, error = %error, "index failed, backing off"); + } + Ok(()) + } } diff --git a/gitmirror/crates/gitmirror-index/Cargo.toml b/gitmirror/crates/gitmirror-index/Cargo.toml new file mode 100644 index 000000000..b45928230 --- /dev/null +++ b/gitmirror/crates/gitmirror-index/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "gitmirror-index" +version.workspace = true +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[dependencies] +bobbin-runtime = { workspace = true } +gitmirror-db = { workspace = true } +gitmirror-git = { workspace = true } +gitmirror-types = { workspace = true } +knot-git = { workspace = true } +knot-langs = { workspace = true } +knot-types = { workspace = true } + +jacquard-common = { workspace = true } +reqwest = { workspace = true } +scc = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true } +tokio-util = { workspace = true } +tracing = { workspace = true } + +[lints] +workspace = true diff --git a/gitmirror/crates/gitmirror-index/src/lib.rs b/gitmirror/crates/gitmirror-index/src/lib.rs new file mode 100644 index 000000000..b54209dd6 --- /dev/null +++ b/gitmirror/crates/gitmirror-index/src/lib.rs @@ -0,0 +1,209 @@ +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bobbin_runtime::Clock; +use gitmirror_db::{Db, DbError, IndexRow}; +use gitmirror_git::layout::Layout; +use gitmirror_types::{backoff_us, now_us}; +use jacquard_common::types::did::Did; +use thiserror::Error; +use tokio::sync::{Notify, Semaphore}; +use tokio_util::sync::CancellationToken; +use tracing::{Instrument as _, debug, error, info, warn}; + +const BATCH_SIZE: i64 = 64; + +#[derive(Clone, Debug)] +pub struct IndexConfig { + pub concurrency: usize, + pub poll_interval: Duration, + pub langs_timeout: Duration, + pub zoekt_url: Option, +} + +pub struct IndexRuntime { + pub db: Db, + pub clock: Arc, + pub cancel: CancellationToken, + pub http: reqwest::Client, + pub layout: Layout, + pub notify: Arc, +} + +#[derive(Debug, Error)] +pub enum IndexError { + #[error("database: {0}")] + Db(#[from] DbError), +} + +struct Job { + db: Db, + clock: Arc, + http: reqwest::Client, + layout: Layout, + langs_timeout: Duration, + zoekt_url: Option, +} + +struct InFlightGuard { + did: Did, + set: Arc>, +} + +impl Drop for InFlightGuard { + fn drop(&mut self) { + let _ = self.set.remove_sync(&self.did); + } +} + +pub async fn run(config: IndexConfig, rt: IndexRuntime) -> Result<(), IndexError> { + info!("index worker started"); + + let job = Arc::new(Job { + db: rt.db.clone(), + clock: rt.clock.clone(), + http: rt.http.clone(), + layout: rt.layout.clone(), + langs_timeout: config.langs_timeout, + zoekt_url: config.zoekt_url.clone(), + }); + let semaphore = Arc::new(Semaphore::new(config.concurrency)); + let in_flight: Arc> = Arc::new(scc::HashSet::new()); + let drained = Arc::new(Notify::new()); + + loop { + if rt.cancel.is_cancelled() { + break; + } + + let mut spawned = 0; + for row in rt.db.pending_index(now_us(&*rt.clock), BATCH_SIZE).await? { + if in_flight.contains_sync(&row.did) { + continue; + } + let Ok(permit) = semaphore.clone().try_acquire_owned() else { + break; + }; + if in_flight.insert_sync(row.did.clone()).is_err() { + continue; + } + let guard = InFlightGuard { + did: row.did.clone(), + set: in_flight.clone(), + }; + + let job = job.clone(); + let drained = drained.clone(); + let span = tracing::info_span!("index", did = %row.did); + tokio::spawn( + async move { + let _guard = guard; + let _permit = permit; + job.run(row).await; + drained.notify_one(); + } + .instrument(span), + ); + spawned += 1; + } + + if spawned == 0 { + tokio::select! { + () = rt.cancel.cancelled() => break, + () = drained.notified() => {} + () = rt.notify.notified() => {} + () = rt.clock.sleep(config.poll_interval) => {} + } + } + } + + info!("index worker stopped"); + Ok(()) +} + +impl Job { + async fn run(&self, row: IndexRow) { + let commit = row.default_branch_commit.as_deref(); + + // trigger zoekt code index + // TODO(post-1.0): zoekt can sync on its own, not relying on here. + if let (Some(url), Some(commit)) = (&self.zoekt_url, commit) { + tokio::spawn( + request_zoekt( + self.http.clone(), + format!("{}/admin/enqueueIndex", url.trim_end_matches('/')), + serde_json::json!({ + "repo": row.did.as_str(), + "branches": [{ "name": row.default_branch.as_str(), "version": commit }], + }), + ) + .in_current_span(), + ); + } + + // TODO: index user commit activity (skip when repo is fork) + // from previous index being `` and updated HEAD being ``, + // 1. discard activity indicies between .. (all commits reachable from + // previous and not from head) + // 2. index user activity between .. (all commits reachable from head and + // not from previous) + + let langs = match commit { + Some(commit) => { + index_languages(self.layout.repo_path(&row.did), commit, self.langs_timeout).await + } + None => Ok(Vec::new()), + }; + let write = match langs { + Ok(langs) => self.db.set_languages(&row.did, commit, &langs).await, + Err(e) => { + self.db + .mark_index_failed( + &row.did, + commit, + &e, + now_us(&*self.clock) + backoff_us(row.index_retry_count), + ) + .await + } + }; + if let Err(e) = write { + error!(error = %e, "recording index outcome failed"); + } + } +} + +async fn index_languages( + path: PathBuf, + commit: &str, + timeout: Duration, +) -> Result, String> { + let commit = knot_types::Oid::from_hex(commit).map_err(|e| e.to_string())?; + tokio::task::spawn_blocking(move || { + let repo = knot_git::Repo::open(path).map_err(|e| e.to_string())?; + let started = Instant::now(); + let sizes = knot_langs::analyze(&repo, commit, Some(started + timeout)) + .map_err(|e| e.to_string())?; + debug!(elapsed = ?started.elapsed(), langs = sizes.len(), "analyzed languages"); + Ok(sizes + .into_iter() + .map(|(name, bytes)| { + ( + name.as_str().to_owned(), + i64::try_from(bytes.get()).unwrap_or(i64::MAX), + ) + }) + .collect()) + }) + .await + .map_err(|_| "language index task panicked".to_owned())? +} + +async fn request_zoekt(http: reqwest::Client, url: String, body: serde_json::Value) { + match http.post(&url).json(&body).send().await { + Ok(resp) if resp.status().is_success() => debug!("requested zoekt index"), + Ok(resp) => warn!(status = %resp.status(), "zoekt index request rejected"), + Err(e) => warn!(error = %e, "zoekt index request failed"), + } +} diff --git a/gitmirror/crates/gitmirror-sync/src/lib.rs b/gitmirror/crates/gitmirror-sync/src/lib.rs index 154ace68a..0f51ebcc9 100644 --- a/gitmirror/crates/gitmirror-sync/src/lib.rs +++ b/gitmirror/crates/gitmirror-sync/src/lib.rs @@ -8,7 +8,7 @@ use bobbin_runtime::Clock; use crate::admission::BackfillAdmission; use gitmirror_db::{Db, DbError}; use gitmirror_git::layout::Layout; -use gitmirror_types::KnotPolicy; +use gitmirror_types::{KnotPolicy, backoff_us, now_us}; use jacquard_common::types::did::Did; use thiserror::Error; use tokio::sync::{Notify, Semaphore}; @@ -32,9 +32,6 @@ pub enum SyncError { /// semaphore is what actually limits work in flight. const BATCH_SIZE: i64 = 64; -/// Ceiling on the exponential retry backoff. -const MAX_BACKOFF_SECS: i64 = 60 * 60; - /// How long a repo is parked when the knot was busy rather than broken. Short, because this is /// not a backoff: it only has to be long enough to get the row out of the scan window. const DEFER_HOLD_US: i64 = 5 * 1_000_000; @@ -72,6 +69,7 @@ pub struct SyncRuntime { pub directory: Arc, pub layout: Layout, pub knot_policy: KnotPolicy, + pub index_notify: Arc, } // NOTE: see hydrant/src/backfill/worker.rs as reference @@ -112,18 +110,6 @@ impl Drop for InFlightGuard { } } -/// The clock reading `retry_after_us` is written from and compared against. -fn now_us(clock: &dyn Clock) -> i64 { - i64::try_from(clock.now_unix_micros().raw()).unwrap_or(i64::MAX) -} - -/// 2^n seconds in micros, capped. -fn backoff_us(retry_count: i32) -> i64 { - let shift = retry_count.clamp(0, 62) as u32; - // cap before scaling; the shift alone would overflow the multiply - (1i64 << shift).min(MAX_BACKOFF_SECS) * 1_000_000 -} - impl BackfillWorker { pub async fn run(self) -> Result<(), SyncError> { info!("backfill worker started"); @@ -170,6 +156,7 @@ impl BackfillWorker { let knot_policy = self.rt.knot_policy; let admission = self.admission.clone(); let drained = self.drained.clone(); + let index_notify = self.rt.index_notify.clone(); let span = tracing::info_span!( "backfill", @@ -193,7 +180,7 @@ impl BackfillWorker { let duration = started.elapsed().as_secs_f64(); let outcome = match &res { - Ok(task::TaskDisposition::Finished) => "finished", + Ok(task::TaskDisposition::Finished {..}) => "finished", Ok(task::TaskDisposition::Deferred) => "deferred", Err(_) => "failed", }; @@ -208,8 +195,16 @@ impl BackfillWorker { ); let write = match &res { - Ok(task::TaskDisposition::Finished) => { - db.mark_synced(&row.did, &row.refs_cid).await + Ok(task::TaskDisposition::Finished { + default_branch_commit, + }) => { + let commit = default_branch_commit.map(|oid| oid.to_string()); + let res = + db.mark_synced(&row.did, &row.refs_cid, commit.as_deref()).await; + if res.is_ok() { + index_notify.notify_one(); + } + res } Ok(task::TaskDisposition::Deferred) => { db.mark_deferred( diff --git a/gitmirror/crates/gitmirror-sync/src/task.rs b/gitmirror/crates/gitmirror-sync/src/task.rs index 8cdf28e8e..48c250752 100644 --- a/gitmirror/crates/gitmirror-sync/src/task.rs +++ b/gitmirror/crates/gitmirror-sync/src/task.rs @@ -20,7 +20,7 @@ const AGENT: &str = concat!("gitmirror/", env!("CARGO_PKG_VERSION")); #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TaskDisposition { - Finished, + Finished { default_branch_commit: Option }, Deferred, } @@ -85,13 +85,20 @@ pub async fn did_task( let head = FullName::try_from(format!("refs/heads/{}", repo.refs.default_branch)) .map_err(|e| BackfillError::MalformedRefs(e.to_string()))?; + let default_branch_commit = refs + .iter() + .find(|(_, name)| *name == head) + .map(|(oid, _)| *oid); + let url = format!("{knot}{}", repo.did.as_str()); let path = layout.repo_path(&repo.did); tokio::task::spawn_blocking(move || fetch_into(&url, &path, &refs, head)) .await .map_err(|_| BackfillError::Panicked)??; - Ok(TaskDisposition::Finished) + Ok(TaskDisposition::Finished { + default_branch_commit, + }) } /// The repo's own knot, from its DID document. Re-resolved every task: a repo can migrate. diff --git a/gitmirror/crates/gitmirror-types/Cargo.toml b/gitmirror/crates/gitmirror-types/Cargo.toml index 63fe541fe..f29853a31 100644 --- a/gitmirror/crates/gitmirror-types/Cargo.toml +++ b/gitmirror/crates/gitmirror-types/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true rust-version.workspace = true [dependencies] +bobbin-runtime = { workspace = true } jacquard-common = { workspace = true } bobbin-knot-proxy = { workspace = true } diff --git a/gitmirror/crates/gitmirror-types/src/lib.rs b/gitmirror/crates/gitmirror-types/src/lib.rs index a94cd7c6c..471edb0e4 100644 --- a/gitmirror/crates/gitmirror-types/src/lib.rs +++ b/gitmirror/crates/gitmirror-types/src/lib.rs @@ -2,6 +2,23 @@ pub mod jacquard_ext; pub use bobbin_knot_proxy::KnotHost; +use bobbin_runtime::Clock; + +/// Ceiling on the exponential retry backoff. +const MAX_BACKOFF_SECS: i64 = 60 * 60; + +/// The clock reading `retry_after_us` is written from and compared against. +pub fn now_us(clock: &dyn Clock) -> i64 { + i64::try_from(clock.now_unix_micros().raw()).unwrap_or(i64::MAX) +} + +/// 2^n seconds in micros, capped. +pub fn backoff_us(retry_count: i32) -> i64 { + let shift = retry_count.clamp(0, 62) as u32; + // cap before scaling; the shift alone would overflow the multiply + (1i64 << shift).min(MAX_BACKOFF_SECS) * 1_000_000 +} + #[derive(Debug, Clone, Copy)] pub struct KnotPolicy { pub allow_private: bool, diff --git a/gitmirror/crates/gitmirror/Cargo.toml b/gitmirror/crates/gitmirror/Cargo.toml index e56831fab..daf2e42ae 100644 --- a/gitmirror/crates/gitmirror/Cargo.toml +++ b/gitmirror/crates/gitmirror/Cargo.toml @@ -13,6 +13,7 @@ path = "src/main.rs" bobbin-runtime = { workspace = true } gitmirror-db = { workspace = true } gitmirror-git = { workspace = true } +gitmirror-index = { workspace = true } gitmirror-ingest = { workspace = true } gitmirror-sync = { workspace = true } gitmirror-types = { workspace = true } diff --git a/gitmirror/crates/gitmirror/src/config.rs b/gitmirror/crates/gitmirror/src/config.rs index 756b8d96d..e6fa5d2bd 100644 --- a/gitmirror/crates/gitmirror/src/config.rs +++ b/gitmirror/crates/gitmirror/src/config.rs @@ -31,6 +31,8 @@ pub struct MirrorConfig { pub ingest: IngestConfig, #[config(nested)] pub sync: SyncConfig, + #[config(nested)] + pub index: IndexConfig, } #[derive(Debug, Config)] @@ -182,6 +184,25 @@ pub struct SyncConfig { pub poll_interval_secs: u64, } +#[derive(Debug, Config)] +pub struct IndexConfig { + /// How many repositories may be indexed at once. + #[config(env = "GITMIRROR_INDEX_CONCURRENCY", default = 4)] + pub concurrency: usize, + + /// Seconds the index worker waits before rescanning the queue when it found nothing to do. + #[config(env = "GITMIRROR_INDEX_POLL_INTERVAL_SECS", default = 5)] + pub poll_interval_secs: u64, + + /// Seconds a language analysis may run before it stops and keeps what it counted so far. + #[config(env = "GITMIRROR_INDEX_LANGS_TIMEOUT_SECS", default = 60)] + pub langs_timeout_secs: u64, + + /// Base URL of the zoekt index server. Code search indexing is skipped when unset. + #[config(env = "GITMIRROR_ZOEKT_URL")] + pub zoekt_url: Option, +} + pub fn load(path: Option<&PathBuf>) -> anyhow::Result { let mut builder = MirrorConfig::builder().env(); if let Some(p) = path { diff --git a/gitmirror/crates/gitmirror/src/main.rs b/gitmirror/crates/gitmirror/src/main.rs index 8b9604ba1..577cee71a 100644 --- a/gitmirror/crates/gitmirror/src/main.rs +++ b/gitmirror/crates/gitmirror/src/main.rs @@ -203,6 +203,8 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { let clock: Arc = Arc::new(SystemClock::new()); let cancel = CancellationToken::new(); + let index_notify = Arc::new(tokio::sync::Notify::new()); + let layout = Layout::new(cfg.repo.scan_path.clone()); let worker = BackfillWorker::new( SyncRuntime { @@ -211,11 +213,12 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { cancel: cancel.clone(), directory, http: http.clone(), - layout: Layout::new(cfg.repo.scan_path.clone()), + layout: layout.clone(), knot_policy: gitmirror_types::KnotPolicy { allow_private: cfg.knot.allow_private, require_https: cfg.knot.require_https, }, + index_notify: index_notify.clone(), }, gitmirror_sync::SyncConfig { concurrency: cfg.sync.concurrency, @@ -227,6 +230,20 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { hydrant_url: cfg.ingest.hydrant_url.clone(), checkpoint_interval: Duration::from_secs(cfg.ingest.checkpoint_interval_secs), }; + let index_cfg = gitmirror_index::IndexConfig { + concurrency: cfg.index.concurrency, + poll_interval: Duration::from_secs(cfg.index.poll_interval_secs), + langs_timeout: Duration::from_secs(cfg.index.langs_timeout_secs), + zoekt_url: cfg.index.zoekt_url.clone(), + }; + let index_runtime = gitmirror_index::IndexRuntime { + db: db.clone(), + clock: clock.clone(), + cancel: cancel.clone(), + http: http.clone(), + layout, + notify: index_notify, + }; let ingest_runtime = IngestRuntime { db: db.clone(), clock: clock.clone(), @@ -239,8 +256,10 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { ("ingest", res.map_err(anyhow::Error::from)) }); tasks.spawn(async move { ("sync", worker.run().await.map_err(anyhow::Error::from)) }); - - // TODO: run indexer + tasks.spawn(async move { + let res = gitmirror_index::run(index_cfg, index_runtime).await; + ("index", res.map_err(anyhow::Error::from)) + }); let max_blob_bytes = cfg.repo.max_blob_bytes; let max_archive_bytes = cfg.repo.max_archive_bytes; -- 2.54.0