mirror of
https://seed.radicle.dev/rad:z4V1sjrXqjvFdnCUbxPFqd5p4DtH5.git
synced 2026-10-18 19:30:27 +08:00
radicle-search: Add search backend and httpd endpoints
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
name: check-radicle-httpd
|
||||
name: check-rust
|
||||
on:
|
||||
push:
|
||||
workflow_dispatch:
|
||||
@@ -10,54 +10,49 @@ jobs:
|
||||
test:
|
||||
name: Build & Test
|
||||
runs-on: ubuntu-latest
|
||||
defaults:
|
||||
run:
|
||||
working-directory: ./radicle-httpd
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
- uses: dtolnay/rust-toolchain@1.87
|
||||
- uses: dtolnay/rust-toolchain@1.95
|
||||
- uses: Swatinem/rust-cache@v2
|
||||
with:
|
||||
workspaces: radicle-httpd -> target
|
||||
- name: Build
|
||||
run: cargo build --all-features
|
||||
workspaces: ". -> target"
|
||||
- name: Build (all features)
|
||||
run: cargo build --workspace --all-features
|
||||
env:
|
||||
RUSTFLAGS: -D warnings
|
||||
- name: Run tests
|
||||
run: cargo test --all-features
|
||||
- name: Test (all features)
|
||||
run: cargo test --workspace --all-features
|
||||
- name: Test httpd without search
|
||||
run: cargo test -p radicle-httpd
|
||||
|
||||
docs:
|
||||
name: Docs
|
||||
runs-on: ubuntu-latest
|
||||
defaults:
|
||||
run:
|
||||
working-directory: ./radicle-httpd
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
- uses: dtolnay/rust-toolchain@1.87
|
||||
- uses: dtolnay/rust-toolchain@1.95
|
||||
- uses: Swatinem/rust-cache@v2
|
||||
with:
|
||||
workspaces: radicle-httpd -> target
|
||||
workspaces: ". -> target"
|
||||
- name: Docs
|
||||
run: cargo doc --all-features
|
||||
run: cargo doc --workspace --all-features
|
||||
env:
|
||||
RUSTDOCFLAGS: -D warnings
|
||||
|
||||
lint:
|
||||
name: Lint
|
||||
runs-on: ubuntu-latest
|
||||
defaults:
|
||||
run:
|
||||
working-directory: ./radicle-httpd
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
- uses: dtolnay/rust-toolchain@1.87
|
||||
- uses: dtolnay/rust-toolchain@1.95
|
||||
with:
|
||||
components: rustfmt, clippy
|
||||
- uses: Swatinem/rust-cache@v2
|
||||
with:
|
||||
workspaces: radicle-httpd -> target
|
||||
workspaces: ". -> target"
|
||||
- name: Run clippy
|
||||
run: cargo clippy --tests
|
||||
run: cargo clippy --workspace --all-features --tests
|
||||
env:
|
||||
RUSTFLAGS: -D warnings
|
||||
- name: Check formatting
|
||||
run: cargo fmt -- --check
|
||||
run: cargo fmt --all -- --check
|
||||
Generated
+1089
-660
File diff suppressed because it is too large
Load Diff
+21
@@ -0,0 +1,21 @@
|
||||
[workspace]
|
||||
members = ["crates/*"]
|
||||
resolver = "2"
|
||||
|
||||
[workspace.package]
|
||||
license = "MIT OR Apache-2.0"
|
||||
homepage = "https://radicle.dev"
|
||||
|
||||
[workspace.dependencies]
|
||||
anyhow = "1"
|
||||
chrono = { version = "0.4.41", default-features = false, features = ["clock"] }
|
||||
lexopt = "0.3.1"
|
||||
meilisearch-sdk = "0.33.0"
|
||||
radicle = "0.24.0"
|
||||
radicle-surf = { version = "0.27.1", default-features = false }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
thiserror = "2"
|
||||
tokio = { version = "1.47.1", default-features = false }
|
||||
tracing = { version = "0.1.41", default-features = false, features = ["std", "log"] }
|
||||
tracing-subscriber = { version = "0.3.19", default-features = false, features = ["std", "ansi", "fmt"] }
|
||||
@@ -1,8 +1,8 @@
|
||||
[package]
|
||||
name = "radicle-httpd"
|
||||
description = "Radicle HTTP daemon"
|
||||
homepage = "https://radicle.dev"
|
||||
license = "MIT OR Apache-2.0"
|
||||
homepage.workspace = true
|
||||
license.workspace = true
|
||||
version = "0.25.0"
|
||||
authors = ["cloudhead <cloudhead@radicle.xyz>", "sebastinez <me@sebastinez.dev>"]
|
||||
edition = "2021"
|
||||
@@ -21,38 +21,39 @@ name = "radicle-httpd"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
anyhow = { version = "1" }
|
||||
anyhow.workspace = true
|
||||
axum = { version = "0.8.4", default-features = false, features = ["json", "query", "tokio", "http1"] }
|
||||
axum-listener = { version = "0.2.2" }
|
||||
base64 = { version = "0.22.1" }
|
||||
chrono = { version = "0.4.41", default-features = false, features = ["clock"] }
|
||||
chrono.workspace = true
|
||||
futures-util = { version = "0.3.32", default-features = false }
|
||||
hyper = { version = "1.6.0", default-features = false }
|
||||
infer = { version = "0.19.0" }
|
||||
lexopt = { version = "0.3.1" }
|
||||
lexopt.workspace = true
|
||||
lru = { version = "0.16.0" }
|
||||
mime_guess = { version = "2.0.5" }
|
||||
nonempty = { version = "0.12.0", features = ["serialize"] }
|
||||
radicle = { version = "0.24.0" }
|
||||
radicle.workspace = true
|
||||
radicle-job = { version = "0.6.0" }
|
||||
radicle-surf = { version = "0.27.1", default-features = false, features = ["serde"] }
|
||||
radicle-search = { path = "../radicle-search" }
|
||||
radicle-surf = { workspace = true, features = ["serde"] }
|
||||
radicle-term = { version = "0.18.0", default-features = false }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = { version = "1", features = ["preserve_order"] }
|
||||
thiserror = { version = "2" }
|
||||
tokio = { version = "1.47.1", default-features = false, features = ["macros", "rt-multi-thread", "signal", "process", "io-util"] }
|
||||
serde.workspace = true
|
||||
serde_json = { workspace = true, features = ["preserve_order"] }
|
||||
thiserror.workspace = true
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal", "process", "io-util"] }
|
||||
tokio-util = { version = "0.7.18", default-features = false, features = ["io"] }
|
||||
tower-http = { version = "0.6.6", default-features = false, features = ["trace", "cors", "set-header", "decompression-full"] }
|
||||
tracing = { version = "0.1.41", default-features = false, features = ["std", "log"] }
|
||||
tracing.workspace = true
|
||||
tracing-logfmt = { version = "0.3.5", optional = true }
|
||||
tracing-subscriber = { version = "0.3.19", default-features = false, features = ["std", "ansi", "fmt"] }
|
||||
tracing-subscriber.workspace = true
|
||||
url = { version = "2.5.4", features = ["serde"] }
|
||||
uuid = { version = "1.13.1", features = ["serde"] }
|
||||
|
||||
[dev-dependencies]
|
||||
hyper = { version = "1.6", default-features = false, features = ["client"] }
|
||||
pretty_assertions = { version = "1.4.1" }
|
||||
radicle = { version = "0.24.0", features = ["test"] }
|
||||
radicle = { workspace = true, features = ["test"] }
|
||||
radicle-crypto = { version = "0.17.0", features = ["test"] }
|
||||
tempfile = { version = "3.20.0" }
|
||||
tower = { version = "0.5.2", features = ["util"] }
|
||||
@@ -21,6 +21,8 @@ mod json;
|
||||
pub(crate) mod query;
|
||||
mod v1;
|
||||
|
||||
pub(crate) use radicle_search::query::SearchClient;
|
||||
|
||||
use crate::api::error::Error;
|
||||
use crate::cache::Cache;
|
||||
use crate::Options;
|
||||
@@ -69,15 +71,59 @@ pub struct Context {
|
||||
profile: Arc<Profile>,
|
||||
cache: Option<Cache>,
|
||||
web_config: WebConfig,
|
||||
/// Search backend client, present only when configured via
|
||||
/// `RADICLE_SEARCH_URL`. When absent, listing and search fall back to
|
||||
/// the storage walk.
|
||||
search: Option<SearchClient>,
|
||||
}
|
||||
|
||||
impl Context {
|
||||
pub fn new(profile: Arc<Profile>, web_config: WebConfig, options: &Options) -> Self {
|
||||
Self {
|
||||
pub fn new(
|
||||
profile: Arc<Profile>,
|
||||
web_config: WebConfig,
|
||||
options: &Options,
|
||||
) -> anyhow::Result<Self> {
|
||||
// Search is optional and resolved at runtime: build a client only
|
||||
// when configured. A construction failure is non-fatal — httpd still
|
||||
// starts and serves everything from the storage walk.
|
||||
let search = match &options.search {
|
||||
Some(cfg) => match SearchClient::new(
|
||||
&cfg.url,
|
||||
cfg.api_key.as_deref(),
|
||||
&cfg.index_name,
|
||||
cfg.query_timeout,
|
||||
) {
|
||||
Ok(client) => {
|
||||
tracing::info!(
|
||||
"search backend enabled: url={} index={}",
|
||||
cfg.url,
|
||||
cfg.index_name
|
||||
);
|
||||
Some(client)
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"failed to construct search backend client, \
|
||||
falling back to storage walk: {e:#}"
|
||||
);
|
||||
None
|
||||
}
|
||||
},
|
||||
None => None,
|
||||
};
|
||||
|
||||
Ok(Self {
|
||||
profile: profile.clone(),
|
||||
cache: options.cache.map(Cache::new),
|
||||
web_config,
|
||||
}
|
||||
search,
|
||||
})
|
||||
}
|
||||
|
||||
/// The search backend client, if one is configured and reachable at
|
||||
/// startup. `None` means listing and search use the storage walk.
|
||||
pub fn search(&self) -> Option<&SearchClient> {
|
||||
self.search.as_ref()
|
||||
}
|
||||
|
||||
#[allow(clippy::result_large_err)]
|
||||
@@ -366,13 +412,13 @@ mod search {
|
||||
|
||||
impl Ord for SearchResult {
|
||||
fn cmp(&self, other: &Self) -> Ordering {
|
||||
match (self.index, other.index) {
|
||||
let primary = match (self.index, other.index) {
|
||||
(0, 0) => self.seeds.cmp(&other.seeds),
|
||||
(0, _) => std::cmp::Ordering::Less,
|
||||
(_, 0) => std::cmp::Ordering::Greater,
|
||||
(ai, bi) if ai == bi => self.seeds.cmp(&other.seeds),
|
||||
(_, _) => self.seeds.cmp(&other.seeds),
|
||||
}
|
||||
(0, _) => Ordering::Less,
|
||||
(_, 0) => Ordering::Greater,
|
||||
_ => self.seeds.cmp(&other.seeds),
|
||||
};
|
||||
primary.then_with(|| self.rid.cmp(&other.rid))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,3 +152,14 @@ impl IntoResponse for Error {
|
||||
(status, body).into_response()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn not_found_maps_to_404() {
|
||||
let response = Error::NotFound.into_response();
|
||||
assert_eq!(response.status(), StatusCode::NOT_FOUND);
|
||||
}
|
||||
}
|
||||
@@ -2,11 +2,22 @@ use serde::{Deserialize, Serialize};
|
||||
|
||||
use radicle::cob::{issue, patch};
|
||||
|
||||
/// Upper bound on caller-supplied `per_page` for paginated list and search
|
||||
/// endpoints. Larger requests are silently clamped to this value to avoid
|
||||
/// unbounded fan-out (e.g. N repository lookups, or N Meili results).
|
||||
pub const MAX_PER_PAGE: usize = 100;
|
||||
|
||||
/// Upper bound on caller-supplied `q` length on search endpoints. Queries
|
||||
/// longer than this are truncated before being forwarded to the search
|
||||
/// backend.
|
||||
pub const MAX_QUERY_LEN: usize = 256;
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PaginationQuery {
|
||||
#[serde(default)]
|
||||
pub show: RepoQuery,
|
||||
pub sort: Option<RepoSort>,
|
||||
pub page: Option<usize>,
|
||||
pub per_page: Option<usize>,
|
||||
}
|
||||
@@ -19,6 +30,15 @@ pub enum RepoQuery {
|
||||
Pinned,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Copy, Default, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub enum RepoSort {
|
||||
#[default]
|
||||
Rid,
|
||||
Activity,
|
||||
Seeding,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RawQuery {
|
||||
@@ -1,4 +1,5 @@
|
||||
mod delegates;
|
||||
mod info;
|
||||
mod node;
|
||||
mod repos;
|
||||
mod stats;
|
||||
@@ -18,6 +19,7 @@ pub fn router(ctx: Context) -> Router {
|
||||
|
||||
let routes = Router::new()
|
||||
.merge(root_router)
|
||||
.merge(info::router(ctx.clone()))
|
||||
.merge(node::router(ctx.clone()))
|
||||
.merge(delegates::router(ctx.clone()))
|
||||
.merge(repos::router(ctx.clone()))
|
||||
@@ -45,6 +47,11 @@ async fn root_handler(State(ctx): State<Context>) -> impl IntoResponse {
|
||||
"rel": "repo",
|
||||
"type": "GET"
|
||||
},
|
||||
{
|
||||
"href": "/info",
|
||||
"rel": "info",
|
||||
"type": "GET"
|
||||
},
|
||||
{
|
||||
"href": "/node",
|
||||
"rel": "node",
|
||||
+3
-2
@@ -7,7 +7,7 @@ use radicle::identity::Did;
|
||||
use radicle::storage::ReadStorage;
|
||||
|
||||
use crate::api::error::Error;
|
||||
use crate::api::query::{PaginationQuery, RepoQuery};
|
||||
use crate::api::query::{PaginationQuery, RepoQuery, MAX_PER_PAGE};
|
||||
use crate::api::Context;
|
||||
use crate::axum_extra::{Path, Query};
|
||||
|
||||
@@ -28,9 +28,10 @@ async fn delegates_repos_handler(
|
||||
show,
|
||||
page,
|
||||
per_page,
|
||||
..
|
||||
} = qs;
|
||||
let page = page.unwrap_or(0);
|
||||
let per_page = per_page.unwrap_or(10);
|
||||
let per_page = per_page.unwrap_or(10).min(MAX_PER_PAGE);
|
||||
let storage = &ctx.profile.storage;
|
||||
let web_config = ctx.web_config().read().await;
|
||||
let pinned = &web_config.pinned;
|
||||
@@ -0,0 +1,70 @@
|
||||
use axum::extract::State;
|
||||
use axum::response::IntoResponse;
|
||||
use axum::routing::get;
|
||||
use axum::Router;
|
||||
use serde::Serialize;
|
||||
|
||||
use crate::api::error::Error;
|
||||
use crate::api::v1::node;
|
||||
use crate::api::Context;
|
||||
use crate::axum_extra::cached_response;
|
||||
|
||||
pub fn router(ctx: Context) -> Router {
|
||||
Router::new()
|
||||
.route("/info", get(info_handler))
|
||||
.with_state(ctx)
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct HttpdInfo {
|
||||
/// Whether an external search index is configured for this httpd.
|
||||
/// Reflects only configuration — `true` does not guarantee the
|
||||
/// backend is currently reachable.
|
||||
search_available: bool,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct Info {
|
||||
node: node::Response,
|
||||
httpd: HttpdInfo,
|
||||
}
|
||||
|
||||
/// Aggregate state about this httpd plus the node it sits in front of.
|
||||
/// Allows clients to bootstrap with a single request instead of hitting
|
||||
/// `/node` plus any future per-feature endpoints.
|
||||
/// `GET /info`
|
||||
async fn info_handler(State(ctx): State<Context>) -> impl IntoResponse {
|
||||
let node = node::build_response(&ctx).await?;
|
||||
let httpd = HttpdInfo {
|
||||
search_available: ctx.search().is_some(),
|
||||
};
|
||||
Ok::<_, Error>(cached_response(Info { node, httpd }, 600))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod routes {
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use axum::extract::connect_info::MockConnectInfo;
|
||||
use axum::http::StatusCode;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::test::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_info_without_search() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let seed = seed(tmp.path());
|
||||
let app = super::router(seed.clone())
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let response = get(&app, "/info").await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: Value = response.json().await;
|
||||
assert_eq!(body["httpd"]["searchAvailable"], false);
|
||||
assert!(body["node"].is_object());
|
||||
}
|
||||
}
|
||||
@@ -35,7 +35,7 @@ pub fn router(ctx: Context) -> Router {
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct Response {
|
||||
pub(crate) struct Response {
|
||||
id: String,
|
||||
agent: Option<UserAgent>,
|
||||
config: Option<Config>,
|
||||
@@ -68,9 +68,9 @@ impl Response {
|
||||
}
|
||||
}
|
||||
|
||||
/// Return local node information.
|
||||
/// `GET /node`
|
||||
async fn node_handler(State(ctx): State<Context>) -> impl IntoResponse {
|
||||
/// Build the local node response. Used by `GET /node` and by the
|
||||
/// `GET /info` aggregate.
|
||||
pub(crate) async fn build_response(ctx: &Context) -> Result<Response, Error> {
|
||||
let node_id = ctx.profile.public_key;
|
||||
let home = ctx.profile.database()?;
|
||||
let agent = AddressStore::get(&home, &node_id)
|
||||
@@ -79,7 +79,7 @@ async fn node_handler(State(ctx): State<Context>) -> impl IntoResponse {
|
||||
|
||||
// The call to `is_running` is a blocking call, which has been, anecdotally, slow to respond.
|
||||
// Spawn a thread with a timeout to ensure that the call to `is_running` does not slow down the
|
||||
// response of the `/node` route too much.
|
||||
// response of the route too much.
|
||||
let node_state = {
|
||||
let socket = ctx.profile.socket_from_env();
|
||||
let is_running = timeout(
|
||||
@@ -116,14 +116,19 @@ async fn node_handler(State(ctx): State<Context>) -> impl IntoResponse {
|
||||
}
|
||||
};
|
||||
|
||||
let response = Response::new(
|
||||
Ok(Response::new(
|
||||
node_id,
|
||||
agent,
|
||||
config,
|
||||
node_state.to_string(),
|
||||
ctx.web_config().read().await,
|
||||
);
|
||||
))
|
||||
}
|
||||
|
||||
/// Return local node information.
|
||||
/// `GET /node`
|
||||
async fn node_handler(State(ctx): State<Context>) -> impl IntoResponse {
|
||||
let response = build_response(&ctx).await?;
|
||||
Ok::<_, Error>(cached_response(response, 600))
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
mod job;
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet, HashMap};
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
|
||||
use axum::extract::{DefaultBodyLimit, State};
|
||||
use axum::http::header;
|
||||
@@ -17,12 +17,12 @@ use serde_json::json;
|
||||
use radicle::cob::{issue::cache::Issues as _, patch::cache::Patches as _};
|
||||
use radicle::identity::RepoId;
|
||||
use radicle::node::{Alias, AliasStore, NodeId};
|
||||
use radicle::storage::{ReadRepository, ReadStorage, RemoteRepository};
|
||||
use radicle::storage::{ReadRepository, RemoteRepository};
|
||||
|
||||
use crate::api;
|
||||
use crate::api::error::Error;
|
||||
use crate::api::query::{CobsQuery, PaginationQuery, RepoQuery};
|
||||
use crate::api::search::{SearchQueryString, SearchResult};
|
||||
use crate::api::query::{CobsQuery, PaginationQuery, RepoQuery, MAX_PER_PAGE, MAX_QUERY_LEN};
|
||||
use crate::api::search::SearchQueryString;
|
||||
use crate::api::Context;
|
||||
use crate::api::PeelToCommit;
|
||||
use crate::axum_extra::{cached_response, immutable_response, Path, Query};
|
||||
@@ -54,6 +54,295 @@ pub fn router(ctx: Context) -> Router {
|
||||
.layer(DefaultBodyLimit::max(MAX_BODY_LIMIT))
|
||||
}
|
||||
|
||||
/// Storage-walk implementations for repo listing and search. Used directly
|
||||
/// when no search backend is configured, and as a fallback for sort modes
|
||||
/// the index doesn't serve (pinned, rid) or when a configured backend is
|
||||
/// unreachable.
|
||||
mod storage {
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use radicle::node::routing::Store as _;
|
||||
use radicle::storage::{ReadRepository, ReadStorage};
|
||||
use radicle_surf::Repository;
|
||||
|
||||
use crate::api;
|
||||
use crate::api::error::Error;
|
||||
use crate::api::query::{RepoQuery, RepoSort};
|
||||
use crate::api::search::SearchResult;
|
||||
use crate::api::Context;
|
||||
use crate::axum_extra::cached_response;
|
||||
|
||||
/// Repo listing via storage walk. Activity/seeding sorts are collapsed
|
||||
/// to rid sort — walking storage for every repo per request is too
|
||||
/// expensive without a pre-computed index.
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub fn list_repos(
|
||||
ctx: &Context,
|
||||
show: RepoQuery,
|
||||
sort: RepoSort,
|
||||
page: usize,
|
||||
per_page: usize,
|
||||
web_config: &radicle::web::Config,
|
||||
) -> Result<Vec<api::repo::Info>, Error> {
|
||||
let sort = if matches!(show, RepoQuery::All)
|
||||
&& matches!(sort, RepoSort::Activity | RepoSort::Seeding)
|
||||
{
|
||||
RepoSort::Rid
|
||||
} else {
|
||||
sort
|
||||
};
|
||||
|
||||
let storage = &ctx.profile.storage;
|
||||
let pinned = &web_config.pinned;
|
||||
let policies = ctx.profile.policies()?;
|
||||
|
||||
let repos = match show {
|
||||
RepoQuery::All => storage
|
||||
.repositories()?
|
||||
.into_iter()
|
||||
.filter(|repo| repo.doc.visibility().is_public())
|
||||
.collect::<Vec<_>>(),
|
||||
RepoQuery::Pinned => storage
|
||||
.repositories_by_id(pinned.repositories.iter())
|
||||
.filter_map(|result| match result {
|
||||
Ok(repo) => Some(repo),
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to load pinned repository: {}", e);
|
||||
None
|
||||
}
|
||||
})
|
||||
.filter(|repo| repo.doc.visibility().is_public())
|
||||
.collect::<Vec<_>>(),
|
||||
};
|
||||
|
||||
let infos = match sort {
|
||||
RepoSort::Rid => {
|
||||
let mut repos = repos;
|
||||
repos.sort_by_key(|p| p.rid);
|
||||
repos
|
||||
.into_iter()
|
||||
.filter_map(|info| {
|
||||
if !policies.is_seeding(&info.rid).unwrap_or_default() {
|
||||
return None;
|
||||
}
|
||||
let (repo, doc) = ctx.repo(info.rid).ok()?;
|
||||
ctx.repo_info(&repo, doc).ok()
|
||||
})
|
||||
.skip(page * per_page)
|
||||
.take(per_page)
|
||||
.collect::<Vec<_>>()
|
||||
}
|
||||
RepoSort::Activity => {
|
||||
let mut with_time: Vec<(radicle::identity::RepoId, i64)> = repos
|
||||
.into_iter()
|
||||
.filter_map(|info| {
|
||||
if !policies.is_seeding(&info.rid).unwrap_or_default() {
|
||||
return None;
|
||||
}
|
||||
let (repo, _doc) = ctx.repo(info.rid).ok()?;
|
||||
let surf = Repository::open(repo.path()).ok()?;
|
||||
let head = surf.head().ok()?;
|
||||
let commit = surf.commit(head).ok()?;
|
||||
Some((info.rid, commit.committer.time.seconds()))
|
||||
})
|
||||
.collect();
|
||||
with_time.sort_by_key(|x| std::cmp::Reverse(x.1));
|
||||
with_time
|
||||
.into_iter()
|
||||
.skip(page * per_page)
|
||||
.take(per_page)
|
||||
.filter_map(|(rid, _)| {
|
||||
let (repo, doc) = ctx.repo(rid).ok()?;
|
||||
ctx.repo_info(&repo, doc).ok()
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
}
|
||||
RepoSort::Seeding => {
|
||||
let db = ctx.profile.database()?;
|
||||
let mut with_count: Vec<(radicle::identity::RepoId, usize)> = repos
|
||||
.into_iter()
|
||||
.filter_map(|info| {
|
||||
if !policies.is_seeding(&info.rid).unwrap_or_default() {
|
||||
return None;
|
||||
}
|
||||
let count = db.count(&info.rid).unwrap_or_default();
|
||||
Some((info.rid, count))
|
||||
})
|
||||
.collect();
|
||||
with_count.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
|
||||
with_count
|
||||
.into_iter()
|
||||
.skip(page * per_page)
|
||||
.take(per_page)
|
||||
.filter_map(|(rid, _)| {
|
||||
let (repo, doc) = ctx.repo(rid).ok()?;
|
||||
ctx.repo_info(&repo, doc).ok()
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
}
|
||||
};
|
||||
|
||||
Ok(infos)
|
||||
}
|
||||
|
||||
/// Substring search via storage walk. Matches are sorted by position
|
||||
/// of the first match in the name (prefix matches first), with seeding
|
||||
/// count as tie-breaker. Results are cached for 10 minutes.
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub async fn search_repos(
|
||||
ctx: &Context,
|
||||
q: &str,
|
||||
page: usize,
|
||||
per_page: usize,
|
||||
) -> Result<axum::response::Response, Error> {
|
||||
use axum::response::IntoResponse;
|
||||
|
||||
let storage = &ctx.profile.storage;
|
||||
let aliases = &ctx.profile.aliases();
|
||||
let db = &ctx.profile.database()?;
|
||||
let found_repos = storage
|
||||
.repositories()?
|
||||
.into_iter()
|
||||
.filter_map(|info| SearchResult::new(q, info, db, aliases))
|
||||
.collect::<BTreeSet<SearchResult>>();
|
||||
|
||||
let found_repos = found_repos
|
||||
.into_iter()
|
||||
.skip(page * per_page)
|
||||
.take(per_page)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
Ok(cached_response(found_repos, 600).into_response())
|
||||
}
|
||||
}
|
||||
|
||||
/// Repo listing and search dispatch. When a search backend is configured
|
||||
/// (`ctx.search()` is `Some`), activity/seeding sorts and `/repos/search` are
|
||||
/// served from the Meilisearch index, transparently falling back to the
|
||||
/// storage walk on any backend failure. Without a backend, or for sort modes
|
||||
/// the index doesn't serve (pinned, rid), everything uses the storage walk.
|
||||
mod listing {
|
||||
use axum::response::IntoResponse;
|
||||
use axum::Json;
|
||||
use radicle::node::routing::Store as _;
|
||||
use radicle::node::AliasStore;
|
||||
use radicle_search::query::{SearchClient, SortField};
|
||||
use serde_json::json;
|
||||
|
||||
use crate::api;
|
||||
use crate::api::error::Error;
|
||||
use crate::api::query::{RepoQuery, RepoSort};
|
||||
use crate::api::search::SearchResult;
|
||||
use crate::api::Context;
|
||||
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub async fn list_repos(
|
||||
ctx: &Context,
|
||||
show: RepoQuery,
|
||||
sort: RepoSort,
|
||||
page: usize,
|
||||
per_page: usize,
|
||||
web_config: &radicle::web::Config,
|
||||
) -> Result<Vec<api::repo::Info>, Error> {
|
||||
let search_sort = match (&show, sort) {
|
||||
(RepoQuery::All, RepoSort::Activity) => Some(SortField::HeadCommitterTime),
|
||||
(RepoQuery::All, RepoSort::Seeding) => Some(SortField::SeedingCount),
|
||||
_ => None,
|
||||
};
|
||||
if let (Some(field), Some(client)) = (search_sort, ctx.search()) {
|
||||
match sorted_repos(ctx, client, field, page, per_page).await {
|
||||
Ok(infos) => return Ok(infos),
|
||||
Err(e) => {
|
||||
tracing::warn!("search backend failed, falling back to storage walk ({e:#})")
|
||||
}
|
||||
}
|
||||
}
|
||||
super::storage::list_repos(ctx, show, sort, page, per_page, web_config)
|
||||
}
|
||||
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub async fn search_repos(
|
||||
ctx: &Context,
|
||||
q: &str,
|
||||
page: usize,
|
||||
per_page: usize,
|
||||
) -> Result<axum::response::Response, Error> {
|
||||
if let Some(client) = ctx.search() {
|
||||
match search_by_query(ctx, client, q, page, per_page).await {
|
||||
Ok(response) => return Ok(response),
|
||||
Err(e) => {
|
||||
tracing::warn!("search backend failed, falling back to storage walk ({e:#})")
|
||||
}
|
||||
}
|
||||
}
|
||||
super::storage::search_repos(ctx, q, page, per_page).await
|
||||
}
|
||||
|
||||
/// Full-text search via Meilisearch. Typo-tolerant prefix matching with
|
||||
/// seedingCount tie-breaking. Returns `Err` on any backend failure so the
|
||||
/// caller falls back to the storage walk.
|
||||
async fn search_by_query(
|
||||
ctx: &Context,
|
||||
client: &SearchClient,
|
||||
q: &str,
|
||||
page: usize,
|
||||
per_page: usize,
|
||||
) -> anyhow::Result<axum::response::Response> {
|
||||
let rids = client.search_by_query(q, page * per_page, per_page).await?;
|
||||
let aliases = ctx.profile.aliases();
|
||||
let db = ctx.profile.database()?;
|
||||
let found_repos: Vec<SearchResult> = rids
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.filter_map(|(i, rid)| {
|
||||
let (_repo, doc_at) = ctx.repo(rid).ok()?;
|
||||
let seeds = db.count(&rid).unwrap_or_default();
|
||||
let delegates = doc_at
|
||||
.doc
|
||||
.delegates()
|
||||
.iter()
|
||||
.map(|did| match aliases.alias(did) {
|
||||
Some(alias) => json!({ "id": did, "alias": alias }),
|
||||
None => json!({ "id": did }),
|
||||
})
|
||||
.collect();
|
||||
Some(SearchResult {
|
||||
rid,
|
||||
payloads: doc_at.doc.payload().clone(),
|
||||
delegates,
|
||||
seeds,
|
||||
index: i,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
Ok(Json(found_repos).into_response())
|
||||
}
|
||||
|
||||
/// Sorted repo listing (activity/seeding) via the index. Returns `Err` on
|
||||
/// backend failure so the caller falls back to the storage walk.
|
||||
async fn sorted_repos(
|
||||
ctx: &Context,
|
||||
client: &SearchClient,
|
||||
field: SortField,
|
||||
page: usize,
|
||||
per_page: usize,
|
||||
) -> anyhow::Result<Vec<api::repo::Info>> {
|
||||
let policies = ctx.profile.policies()?;
|
||||
let offset = page.saturating_mul(per_page);
|
||||
let rids = client.sorted_rids(field, offset, per_page).await?;
|
||||
Ok(rids
|
||||
.into_iter()
|
||||
.filter_map(|rid| {
|
||||
if !policies.is_seeding(&rid).unwrap_or_default() {
|
||||
return None;
|
||||
}
|
||||
let (repo, doc) = ctx.repo(rid).ok()?;
|
||||
ctx.repo_info(&repo, doc).ok()
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
}
|
||||
|
||||
/// List all repos.
|
||||
/// `GET /repos`
|
||||
async fn repo_root_handler(
|
||||
@@ -62,95 +351,36 @@ async fn repo_root_handler(
|
||||
) -> impl IntoResponse {
|
||||
let PaginationQuery {
|
||||
show,
|
||||
sort,
|
||||
page,
|
||||
per_page,
|
||||
} = qs;
|
||||
let page = page.unwrap_or(0);
|
||||
let sort = sort.unwrap_or_default();
|
||||
let web_config = ctx.web_config().read().await;
|
||||
let per_page = per_page.unwrap_or_else(|| match show {
|
||||
RepoQuery::Pinned => web_config.pinned.repositories.len(),
|
||||
_ => 10,
|
||||
});
|
||||
let storage = &ctx.profile.storage;
|
||||
let pinned = &web_config.pinned;
|
||||
let policies = ctx.profile.policies()?;
|
||||
|
||||
let mut repos = match show {
|
||||
RepoQuery::All => storage
|
||||
.repositories()?
|
||||
.into_iter()
|
||||
.filter(|repo| repo.doc.visibility().is_public())
|
||||
.collect::<Vec<_>>(),
|
||||
RepoQuery::Pinned => storage
|
||||
.repositories_by_id(pinned.repositories.iter())
|
||||
.filter_map(|result| match result {
|
||||
Ok(repo) => Some(repo),
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to load pinned repository: {}", e);
|
||||
None
|
||||
}
|
||||
})
|
||||
.filter(|repo| repo.doc.visibility().is_public())
|
||||
.collect::<Vec<_>>(),
|
||||
let per_page = match per_page {
|
||||
Some(n) => n.min(MAX_PER_PAGE),
|
||||
None => match show {
|
||||
RepoQuery::Pinned => web_config.pinned.repositories.len(),
|
||||
_ => 10,
|
||||
},
|
||||
};
|
||||
repos.sort_by_key(|p| p.rid);
|
||||
|
||||
let infos = repos
|
||||
.into_iter()
|
||||
.filter_map(|info| {
|
||||
if !policies.is_seeding(&info.rid).unwrap_or_default() {
|
||||
return None;
|
||||
}
|
||||
let Ok((repo, doc)) = ctx.repo(info.rid) else {
|
||||
return None;
|
||||
};
|
||||
let Ok(repo_info) = ctx.repo_info(&repo, doc) else {
|
||||
return None;
|
||||
};
|
||||
|
||||
Some(repo_info)
|
||||
})
|
||||
.skip(page * per_page)
|
||||
.take(per_page)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let infos = listing::list_repos(&ctx, show, sort, page, per_page, &web_config).await?;
|
||||
Ok::<_, Error>(Json(infos))
|
||||
}
|
||||
|
||||
/// Search repositories by name.
|
||||
/// `GET /repos/search?q=<query>`
|
||||
///
|
||||
/// We obtain the byte index of the first character of the query that matches the repo name.
|
||||
/// And skip if the query doesn't match the repo name.
|
||||
///
|
||||
/// Sorting algorithm:
|
||||
/// If both byte indices are 0, compare by seeding count.
|
||||
/// A repo name with a byte index of 0 should come before non-zero indices.
|
||||
/// If both indices are non-zero and equal, then compare by seeding count.
|
||||
/// If none of the above, all non-zero indices are compared by their seeding count primarily.
|
||||
async fn repo_search_handler(
|
||||
State(ctx): State<Context>,
|
||||
Query(SearchQueryString { q, per_page, page }): Query<SearchQueryString>,
|
||||
) -> impl IntoResponse {
|
||||
let q = q.unwrap_or_default();
|
||||
let q: String = q.unwrap_or_default().chars().take(MAX_QUERY_LEN).collect();
|
||||
let page = page.unwrap_or(0);
|
||||
let per_page = per_page.unwrap_or(10);
|
||||
let storage = &ctx.profile.storage;
|
||||
let aliases = &ctx.profile.aliases();
|
||||
let db = &ctx.profile.database()?;
|
||||
let found_repos = storage
|
||||
.repositories()?
|
||||
.into_iter()
|
||||
.filter_map(|info| SearchResult::new(&q, info, db, aliases))
|
||||
.collect::<BTreeSet<SearchResult>>();
|
||||
let per_page = per_page.unwrap_or(10).min(MAX_PER_PAGE);
|
||||
|
||||
let found_repos = found_repos
|
||||
.into_iter()
|
||||
.skip(page * per_page)
|
||||
.take(per_page)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
Ok::<_, Error>(cached_response(found_repos, 600).into_response())
|
||||
listing::search_repos(&ctx, &q, page, per_page).await
|
||||
}
|
||||
|
||||
/// Get repo metadata.
|
||||
@@ -690,7 +920,7 @@ async fn issues_handler(
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
issues.sort_by(|(_, a), (_, b)| b.timestamp().cmp(&a.timestamp()));
|
||||
issues.sort_by_key(|(_, b)| std::cmp::Reverse(b.timestamp()));
|
||||
let aliases = &ctx.profile.aliases();
|
||||
let issues = issues
|
||||
.into_iter()
|
||||
@@ -745,7 +975,7 @@ async fn patches_handler(
|
||||
(status.matches(patch.state())).then_some((id, patch))
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
patches.sort_by(|(_, a), (_, b)| b.timestamp().cmp(&a.timestamp()));
|
||||
patches.sort_by_key(|(_, b)| std::cmp::Reverse(b.timestamp()));
|
||||
let aliases = ctx.profile.aliases();
|
||||
let patches = patches
|
||||
.into_iter()
|
||||
@@ -964,6 +1194,136 @@ mod routes {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_root_sort_seeding() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let response = get(&app, "/repos?show=all&sort=seeding").await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: serde_json::Value = response.json().await;
|
||||
let rids: Vec<&str> = body
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|repo| repo["rid"].as_str().unwrap())
|
||||
.collect();
|
||||
assert_eq!(rids.len(), 2);
|
||||
assert!(rids.contains(&RID));
|
||||
assert!(rids.contains(&"rad:z4GypKmh1gkEfmkXtarcYnkvtFUfE"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_root_sort_activity() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let response = get(&app, "/repos?show=all&sort=activity").await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: serde_json::Value = response.json().await;
|
||||
let rids: Vec<&str> = body
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|repo| repo["rid"].as_str().unwrap())
|
||||
.collect();
|
||||
// Without a search backend, activity sort collapses to rid sort.
|
||||
// hello-world's rid sorts before again-hello-world's.
|
||||
assert_eq!(rids, vec![RID, "rad:z4GypKmh1gkEfmkXtarcYnkvtFUfE"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_root_sort_falls_back_without_search() {
|
||||
// Without a search backend, sort=activity and sort=seeding should
|
||||
// produce the same response as the default rid sort, instead of
|
||||
// walking storage and opening every repo per request.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
|
||||
let rid_sorted: serde_json::Value = get(&app, "/repos?show=all").await.json().await;
|
||||
let activity_sorted: serde_json::Value = get(&app, "/repos?show=all&sort=activity")
|
||||
.await
|
||||
.json()
|
||||
.await;
|
||||
let seeding_sorted: serde_json::Value =
|
||||
get(&app, "/repos?show=all&sort=seeding").await.json().await;
|
||||
|
||||
assert_eq!(rid_sorted, activity_sorted);
|
||||
assert_eq!(rid_sorted, seeding_sorted);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_per_page_is_clamped() {
|
||||
// A caller requesting more than MAX_PER_PAGE must not be able to
|
||||
// pull an unbounded result set. With the small fixture we can't
|
||||
// observe the cap directly, but the request must succeed and the
|
||||
// returned length must respect the cap.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let response = get(&app, "/repos?show=all&perPage=99999").await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: serde_json::Value = response.json().await;
|
||||
let len = body.as_array().unwrap().len();
|
||||
assert!(len <= super::MAX_PER_PAGE);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_search_per_page_is_clamped() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let response = get(&app, "/repos/search?q=hello&perPage=99999").await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: serde_json::Value = response.json().await;
|
||||
let len = body.as_array().unwrap().len();
|
||||
assert!(len <= super::MAX_PER_PAGE);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_search_fallback_sets_cache_control() {
|
||||
// Without a search backend, /repos/search runs the storage-walk
|
||||
// substring scan and must still set Cache-Control so intermediaries
|
||||
// can serve repeat queries from cache.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let response = get(&app, "/repos/search?q=hello").await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let cache_control = response
|
||||
.headers()
|
||||
.get("cache-control")
|
||||
.expect("cache-control header should be present")
|
||||
.to_str()
|
||||
.unwrap();
|
||||
assert!(
|
||||
cache_control.contains("max-age=600"),
|
||||
"unexpected cache-control: {cache_control}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos_search_clamps_long_query() {
|
||||
// A user-supplied `q` longer than MAX_QUERY_LEN is truncated rather
|
||||
// than rejected; the request still succeeds.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let app = super::router(seed(tmp.path()))
|
||||
.layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 8080))));
|
||||
let long_q = "a".repeat(super::MAX_QUERY_LEN * 4);
|
||||
let response = get(&app, format!("/repos/search?q={long_q}")).await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body: serde_json::Value = response.json().await;
|
||||
// Substring scan over "aaa…" finds no repo names; clamp didn't crash.
|
||||
assert_eq!(body.as_array().unwrap().len(), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_repos() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
@@ -51,6 +51,17 @@ pub struct Options {
|
||||
pub aliases: HashMap<String, RepoId>,
|
||||
pub listen: DualAddr,
|
||||
pub cache: Option<NonZeroUsize>,
|
||||
/// Search backend configuration. `None` disables search at runtime and
|
||||
/// falls back to the built-in storage walk.
|
||||
pub search: Option<SearchOptions>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SearchOptions {
|
||||
pub url: String,
|
||||
pub api_key: Option<String>,
|
||||
pub index_name: String,
|
||||
pub query_timeout: std::time::Duration,
|
||||
}
|
||||
|
||||
/// Run the Server.
|
||||
@@ -73,7 +84,7 @@ pub async fn run(options: Options) -> anyhow::Result<()> {
|
||||
|
||||
let web_config = api::WebConfig::from_profile(&profile);
|
||||
let profile = Arc::new(profile);
|
||||
let ctx = api::Context::new(profile.clone(), web_config.clone(), &options);
|
||||
let ctx = api::Context::new(profile.clone(), web_config.clone(), &options)?;
|
||||
|
||||
#[cfg(unix)]
|
||||
tokio::spawn(async move {
|
||||
@@ -251,11 +262,12 @@ mod routes {
|
||||
aliases: HashMap::new(),
|
||||
listen: DualAddr::Tcp(SocketAddr::from(([0, 0, 0, 0], 8080))),
|
||||
cache: None,
|
||||
search: None,
|
||||
};
|
||||
let profile = test::profile(tmp.path(), [0xff; 32]);
|
||||
let web_config = crate::api::WebConfig::from_profile(&profile);
|
||||
let profile = std::sync::Arc::new(profile);
|
||||
let ctx = crate::api::Context::new(profile.clone(), web_config, &options);
|
||||
let ctx = crate::api::Context::new(profile.clone(), web_config, &options).unwrap();
|
||||
let app = super::router(options, profile, ctx)
|
||||
.unwrap()
|
||||
.layer(MockConnectInfo(DualAddr::Tcp(SocketAddr::from((
|
||||
@@ -0,0 +1,218 @@
|
||||
use std::num::NonZeroUsize;
|
||||
use std::{collections::HashMap, process};
|
||||
|
||||
use anyhow::bail;
|
||||
use axum_listener::DualAddr;
|
||||
use radicle::prelude::RepoId;
|
||||
use radicle::version::Version;
|
||||
use radicle_httpd as httpd;
|
||||
|
||||
pub const VERSION: Version = Version {
|
||||
name: "radicle-httpd",
|
||||
commit: env!("GIT_HEAD"),
|
||||
version: env!("RADICLE_VERSION"),
|
||||
timestamp: env!("SOURCE_DATE_EPOCH"),
|
||||
};
|
||||
|
||||
pub const HELP_MSG: &str = r#"
|
||||
Usage
|
||||
|
||||
radicle-httpd [<option>...]
|
||||
|
||||
Options
|
||||
|
||||
--listen <address> Address to listen on: TCP address (e.g., 127.0.0.1:8080)
|
||||
or Unix socket path (e.g., /tmp/radicle.sock)
|
||||
(default: 0.0.0.0:8080)
|
||||
--alias, -a <alias> <rid> Provide alias and RID pairs to shorten git clone commands for repositories,
|
||||
e.g. heartwood and rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5 to produce https://seed.radicle.dev/heartwood.git
|
||||
--cache <number> Max amount of items in cache for /tree endpoints (default: 100)
|
||||
--version, -v Print program version
|
||||
--help, -h Print help
|
||||
"#;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
// SAFETY: The logger is only initialized once.
|
||||
httpd::logger::init().unwrap();
|
||||
tracing::info!("starting http daemon..");
|
||||
tracing::info!("version {} ({})", env!("RADICLE_VERSION"), env!("GIT_HEAD"));
|
||||
|
||||
let options = parse_options()?;
|
||||
|
||||
match httpd::run(options).await {
|
||||
Ok(()) => {}
|
||||
Err(err) => {
|
||||
tracing::error!("Fatal: {:#}", err);
|
||||
process::exit(1);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Parse command-line arguments into HTTP options.
|
||||
fn parse_options() -> anyhow::Result<httpd::Options> {
|
||||
use lexopt::prelude::*;
|
||||
|
||||
let mut parser = lexopt::Parser::from_env();
|
||||
let mut listen = None;
|
||||
let mut aliases = HashMap::new();
|
||||
let mut cache = Some(httpd::DEFAULT_CACHE_SIZE);
|
||||
|
||||
while let Some(arg) = parser.next()? {
|
||||
match arg {
|
||||
Long("listen") => {
|
||||
let addr: DualAddr = parser.value()?.parse()?;
|
||||
|
||||
#[cfg(unix)]
|
||||
// Get socket path and remove it if existing
|
||||
if let DualAddr::Uds(socket_path) = &addr {
|
||||
if let Some(path) = socket_path.as_pathname() {
|
||||
if path.exists() {
|
||||
tracing::info!("Removing existing socket path at {}", path.display());
|
||||
if let Err(e) = std::fs::remove_file(path) {
|
||||
tracing::error!("{e}");
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tracing::error!("Provided socket address isn't a valid path.");
|
||||
process::exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
listen = Some(addr);
|
||||
}
|
||||
Long("alias") | Short('a') => {
|
||||
let alias: String = parser.value()?.parse()?;
|
||||
let id: RepoId = parser.value()?.parse()?;
|
||||
|
||||
aliases.insert(alias, id);
|
||||
}
|
||||
Long("version") | Short('v') => {
|
||||
if let Err(e) = VERSION.write(std::io::stdout()) {
|
||||
tracing::error!("{e}");
|
||||
process::exit(1);
|
||||
};
|
||||
process::exit(0);
|
||||
}
|
||||
Long("cache") => {
|
||||
let size = parser.value()?.parse()?;
|
||||
cache = NonZeroUsize::new(size);
|
||||
}
|
||||
Long("help") | Short('h') => {
|
||||
println!("{HELP_MSG}");
|
||||
process::exit(0);
|
||||
}
|
||||
_ => return Err(arg.unexpected().into()),
|
||||
}
|
||||
}
|
||||
Ok(httpd::Options {
|
||||
aliases,
|
||||
listen: listen.unwrap_or_else(|| DualAddr::Tcp(([0, 0, 0, 0], 8080).into())),
|
||||
cache,
|
||||
search: search_options_from_env()?,
|
||||
})
|
||||
}
|
||||
|
||||
/// Read the search backend configuration from the environment. Search is
|
||||
/// entirely optional and enabled purely at runtime: when `RADICLE_SEARCH_URL`
|
||||
/// is unset (or empty) this returns `None` and httpd serves repo listings and
|
||||
/// search from its built-in storage walk. When set, httpd routes activity/
|
||||
/// seeding sorts and `/repos/search` through the index, transparently falling
|
||||
/// back to the storage walk if the backend is unreachable.
|
||||
fn search_options_from_env() -> anyhow::Result<Option<httpd::SearchOptions>> {
|
||||
let url = match std::env::var("RADICLE_SEARCH_URL") {
|
||||
Ok(url) if !url.is_empty() => url,
|
||||
_ => return Ok(None),
|
||||
};
|
||||
let api_key = std::env::var("RADICLE_SEARCH_KEY")
|
||||
.ok()
|
||||
.filter(|s| !s.is_empty());
|
||||
let index_name = std::env::var("RADICLE_SEARCH_INDEX_NAME")
|
||||
.ok()
|
||||
.filter(|s| !s.is_empty())
|
||||
.unwrap_or_else(|| "repos".to_string());
|
||||
let query_timeout = parse_timeout_ms_from_env("RADICLE_SEARCH_TIMEOUT_MS", 500)?;
|
||||
Ok(Some(httpd::SearchOptions {
|
||||
url,
|
||||
api_key,
|
||||
index_name,
|
||||
query_timeout,
|
||||
}))
|
||||
}
|
||||
|
||||
/// Parse a millisecond duration from an env var. Rejects unparseable values
|
||||
/// and zero (which would make every query time out immediately).
|
||||
fn parse_timeout_ms_from_env(var: &str, default_ms: u64) -> anyhow::Result<std::time::Duration> {
|
||||
let raw = match std::env::var(var) {
|
||||
Ok(v) if !v.is_empty() => v,
|
||||
_ => return Ok(std::time::Duration::from_millis(default_ms)),
|
||||
};
|
||||
let ms: u64 = raw
|
||||
.parse()
|
||||
.map_err(|_| anyhow::anyhow!("{var}={raw:?} is not a non-negative integer"))?;
|
||||
if ms == 0 {
|
||||
bail!("{var}=0 is not allowed (would time out every query immediately)");
|
||||
}
|
||||
Ok(std::time::Duration::from_millis(ms))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::Duration;
|
||||
|
||||
// Process-global env vars: each test uses its own variable name so
|
||||
// parallel runs don't race.
|
||||
|
||||
#[test]
|
||||
fn timeout_unset_returns_default() {
|
||||
let d = parse_timeout_ms_from_env("RADICLE_HTTPD_TEST_TIMEOUT_UNSET", 500).unwrap();
|
||||
assert_eq!(d, Duration::from_millis(500));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeout_empty_returns_default() {
|
||||
let var = "RADICLE_HTTPD_TEST_TIMEOUT_EMPTY";
|
||||
unsafe { std::env::set_var(var, "") };
|
||||
let d = parse_timeout_ms_from_env(var, 250);
|
||||
unsafe { std::env::remove_var(var) };
|
||||
assert_eq!(d.unwrap(), Duration::from_millis(250));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeout_valid_integer_parses() {
|
||||
let var = "RADICLE_HTTPD_TEST_TIMEOUT_VALID";
|
||||
unsafe { std::env::set_var(var, "1500") };
|
||||
let d = parse_timeout_ms_from_env(var, 500);
|
||||
unsafe { std::env::remove_var(var) };
|
||||
assert_eq!(d.unwrap(), Duration::from_millis(1500));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeout_zero_is_rejected() {
|
||||
let var = "RADICLE_HTTPD_TEST_TIMEOUT_ZERO";
|
||||
unsafe { std::env::set_var(var, "0") };
|
||||
let result = parse_timeout_ms_from_env(var, 500);
|
||||
unsafe { std::env::remove_var(var) };
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeout_unparseable_is_rejected() {
|
||||
let var = "RADICLE_HTTPD_TEST_TIMEOUT_BAD";
|
||||
unsafe { std::env::set_var(var, "not-a-number") };
|
||||
let result = parse_timeout_ms_from_env(var, 500);
|
||||
unsafe { std::env::remove_var(var) };
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeout_negative_is_rejected() {
|
||||
let var = "RADICLE_HTTPD_TEST_TIMEOUT_NEG";
|
||||
unsafe { std::env::set_var(var, "-5") };
|
||||
let result = parse_timeout_ms_from_env(var, 500);
|
||||
unsafe { std::env::remove_var(var) };
|
||||
assert!(result.is_err());
|
||||
}
|
||||
}
|
||||
@@ -332,10 +332,11 @@ fn seed_with_signer<G: Signer<Signature>>(
|
||||
aliases: std::collections::HashMap::new(),
|
||||
listen: axum_listener::DualAddr::Tcp(std::net::SocketAddr::from(([0, 0, 0, 0], 8080))),
|
||||
cache: Some(crate::DEFAULT_CACHE_SIZE),
|
||||
search: None,
|
||||
};
|
||||
|
||||
let web_config = crate::api::WebConfig::from_profile(&profile);
|
||||
Context::new(Arc::new(profile), web_config, &options)
|
||||
Context::new(Arc::new(profile), web_config, &options).expect("test context creation failed")
|
||||
}
|
||||
|
||||
/// Create a test context with multiple peers and canonical refs configured.
|
||||
@@ -507,6 +508,10 @@ impl Response {
|
||||
self.0.status()
|
||||
}
|
||||
|
||||
pub fn headers(&self) -> &axum::http::HeaderMap {
|
||||
self.0.headers()
|
||||
}
|
||||
|
||||
pub async fn body(self) -> Bytes {
|
||||
axum::body::to_bytes(self.0.into_body(), usize::MAX)
|
||||
.await
|
||||
@@ -0,0 +1,31 @@
|
||||
[package]
|
||||
name = "radicle-search"
|
||||
description = "Optional search/index daemon for Radicle, alongside radicle-node and radicle-httpd"
|
||||
homepage.workspace = true
|
||||
license.workspace = true
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
default-run = "radicle-search"
|
||||
|
||||
[[bin]]
|
||||
name = "radicle-search"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
chrono.workspace = true
|
||||
lexopt.workspace = true
|
||||
meilisearch-sdk.workspace = true
|
||||
radicle.workspace = true
|
||||
radicle-surf.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
|
||||
tracing.workspace = true
|
||||
tracing-subscriber = { workspace = true, features = ["env-filter"] }
|
||||
|
||||
[dev-dependencies]
|
||||
# `test-util` enables `#[tokio::test(start_paused = true)]` so the retry
|
||||
# backoff tests fast-forward through their sleeps instead of waiting.
|
||||
tokio = { workspace = true, features = ["test-util"] }
|
||||
@@ -0,0 +1,126 @@
|
||||
# radicle-search
|
||||
|
||||
Optional indexing daemon for Radicle. Runs alongside `radicle-node` and
|
||||
maintains a Meilisearch index of repo metadata that `radicle-httpd` can
|
||||
use to serve fast, typo-tolerant repo listings and full-text search.
|
||||
|
||||
When configured, httpd routes `/repos?sort=activity|seeding` and
|
||||
`/repos/search?q=…` through the index. When absent or unreachable, httpd
|
||||
transparently falls back to its built-in storage walk — no configuration
|
||||
on the client side, no API contract changes.
|
||||
|
||||
## What each repository document contains
|
||||
|
||||
The daemon writes one Meilisearch document per repository. Within it, fields
|
||||
play different roles — only the searchable ones are "indexed" in the full-text
|
||||
sense:
|
||||
|
||||
- **Searchable** (matched by `/repos/search?q=…`): `name`, `description`.
|
||||
- **Sortable** (ordering for `/repos?sort=…`): `seedingCount` (most-seeded) and
|
||||
`headCommitterTime` (recently-active).
|
||||
- **Filterable**: `delegates`, `visibility`.
|
||||
- **Stored** — held on the document but not searched, sorted, or filtered:
|
||||
`rid`, head OID, and default branch.
|
||||
|
||||
httpd uses the index to resolve a sorted or matched set of `rid`s, then builds
|
||||
each repo's response (including its activity sparkline) from storage.
|
||||
|
||||
Only public repos this node seeds are indexed.
|
||||
|
||||
## How it stays up to date
|
||||
|
||||
1. **Bootstrap.** On startup and after every event-stream reconnect, the
|
||||
daemon walks the storage tree end-to-end and re-upserts everything.
|
||||
2. **Real time.** Subscribes to the node's control socket (same stream as
|
||||
`rad node events`) and reacts to:
|
||||
- `RefsFetched`, `LocalRefsAnnounced`, `CanonicalRefUpdated`, `RefsSynced`
|
||||
→ re-index the affected repo.
|
||||
- `SeedDiscovered`, `SeedDropped`, `RefsAnnounced` → re-index only if
|
||||
the rid is one we locally seed (filtered against an in-memory cache,
|
||||
so gossip about repos we don't host is dropped at near-zero cost).
|
||||
3. **Periodic rescan.** Every `RADICLE_SEARCH_RESCAN_SECS` (default 1h)
|
||||
as a safety net for missed events.
|
||||
|
||||
## Setup
|
||||
|
||||
### 1. Install Meilisearch
|
||||
|
||||
Follow the official installation guide:
|
||||
<https://www.meilisearch.com/docs/learn/getting_started/installation>
|
||||
|
||||
The simplest path on Linux is the install script:
|
||||
|
||||
```sh
|
||||
curl -L https://install.meilisearch.com | sh
|
||||
```
|
||||
|
||||
This drops a single self-contained `meilisearch` binary in the current
|
||||
directory. Move it onto your `$PATH` (e.g. `~/.local/bin/`).
|
||||
|
||||
Run it:
|
||||
|
||||
```sh
|
||||
meilisearch \
|
||||
--http-addr 127.0.0.1:7700 \
|
||||
--db-path ~/.meilisearch/data \
|
||||
--no-analytics
|
||||
```
|
||||
|
||||
For production set a master key with `--master-key <KEY>` and run with
|
||||
`--env production`; see the Meilisearch docs for hardening guidance.
|
||||
|
||||
### 2. Build and run radicle-search
|
||||
|
||||
`radicle-search` is a member of the Cargo workspace rooted at the
|
||||
repository root (it lives under `crates/`), so build it from there. Cargo
|
||||
places the binary in the workspace-wide `target/`, not under this crate's
|
||||
directory:
|
||||
|
||||
```sh
|
||||
cargo build --release -p radicle-search
|
||||
./target/release/radicle-search
|
||||
```
|
||||
|
||||
The daemon picks up the Radicle profile from `RAD_HOME` and the node
|
||||
control socket from `RAD_SOCKET` (or their defaults). On first launch it
|
||||
bootstraps the index in a few seconds, then listens for node events.
|
||||
|
||||
### 3. Point httpd at the index
|
||||
|
||||
Set `RADICLE_SEARCH_URL` (plus optional friends) before launching httpd:
|
||||
|
||||
```sh
|
||||
RADICLE_SEARCH_URL=http://127.0.0.1:7700 radicle-httpd
|
||||
```
|
||||
|
||||
On boot httpd logs `search backend enabled: url=… index=repos`. With the
|
||||
env var absent, httpd works exactly as before.
|
||||
|
||||
## Configuration
|
||||
|
||||
All via environment variables:
|
||||
|
||||
| Variable | Default | Purpose |
|
||||
|---|---|---|
|
||||
| `RADICLE_SEARCH_MEILI_URL` | `http://localhost:7700` | Meilisearch instance to connect to. |
|
||||
| `RADICLE_SEARCH_MEILI_KEY` | _(none)_ | Meilisearch master key (production mode). |
|
||||
| `RADICLE_SEARCH_INDEX_NAME` | `repos` | Index name this daemon maintains. |
|
||||
| `RADICLE_SEARCH_RESCAN_SECS` | `3600` | Interval between safety-net full rescans. |
|
||||
| `RADICLE_SEARCH_RECONNECT_BACKOFF_SECS` | `5` | Delay before reconnecting after an event-stream disconnect. |
|
||||
| `RAD_HOME` | `~/.radicle` | Standard Radicle profile path. |
|
||||
| `RAD_SOCKET` | `$RAD_HOME/node/control.sock` | Standard node control socket. |
|
||||
|
||||
httpd reads a parallel set to decide, at runtime, whether to use the index:
|
||||
|
||||
| Variable | Default | Purpose |
|
||||
|---|---|---|
|
||||
| `RADICLE_SEARCH_URL` | _(none)_ | Meilisearch instance httpd queries. **Unset disables search** — httpd serves listings and search from its storage walk. |
|
||||
| `RADICLE_SEARCH_KEY` | _(none)_ | Meilisearch API key. |
|
||||
| `RADICLE_SEARCH_INDEX_NAME` | `repos` | Index name to query. |
|
||||
| `RADICLE_SEARCH_TIMEOUT_MS` | `500` | Per-query timeout in milliseconds (must be a non-zero integer). |
|
||||
|
||||
Use the same URL/key/index values in both processes. A single
|
||||
`radicle-httpd` binary handles both modes — no build-time feature flag is
|
||||
involved. When `RADICLE_SEARCH_URL` is unset, or when a query to the
|
||||
backend fails or times out, httpd transparently falls back to the storage
|
||||
walk, so the API behaves identically either way.
|
||||
@@ -0,0 +1,117 @@
|
||||
use std::env;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
|
||||
const DEFAULT_MEILI_URL: &str = "http://localhost:7700";
|
||||
const DEFAULT_INDEX_NAME: &str = "repos";
|
||||
const DEFAULT_RESCAN_INTERVAL_SECS: u64 = 3600;
|
||||
const DEFAULT_RECONNECT_BACKOFF_SECS: u64 = 5;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Config {
|
||||
pub meili_url: String,
|
||||
pub meili_key: Option<String>,
|
||||
pub index_name: String,
|
||||
pub rescan_interval: Duration,
|
||||
pub reconnect_backoff: Duration,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
pub fn from_env() -> Result<Self> {
|
||||
let meili_url =
|
||||
env::var("RADICLE_SEARCH_MEILI_URL").unwrap_or_else(|_| DEFAULT_MEILI_URL.to_string());
|
||||
let meili_key = env::var("RADICLE_SEARCH_MEILI_KEY")
|
||||
.ok()
|
||||
.filter(|k| !k.is_empty());
|
||||
let index_name = env::var("RADICLE_SEARCH_INDEX_NAME")
|
||||
.unwrap_or_else(|_| DEFAULT_INDEX_NAME.to_string());
|
||||
let rescan_interval =
|
||||
duration_from_env("RADICLE_SEARCH_RESCAN_SECS", DEFAULT_RESCAN_INTERVAL_SECS)?;
|
||||
let reconnect_backoff = duration_from_env(
|
||||
"RADICLE_SEARCH_RECONNECT_BACKOFF_SECS",
|
||||
DEFAULT_RECONNECT_BACKOFF_SECS,
|
||||
)?;
|
||||
|
||||
Ok(Self {
|
||||
meili_url,
|
||||
meili_key,
|
||||
index_name,
|
||||
rescan_interval,
|
||||
reconnect_backoff,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn duration_from_env(var: &str, default_secs: u64) -> Result<Duration> {
|
||||
let raw = match env::var(var) {
|
||||
Ok(v) if !v.is_empty() => v,
|
||||
_ => return Ok(Duration::from_secs(default_secs)),
|
||||
};
|
||||
let secs: u64 = raw
|
||||
.parse()
|
||||
.with_context(|| format!("{var}={raw:?} is not a non-negative integer"))?;
|
||||
if secs == 0 {
|
||||
bail!("{var}=0 is not allowed (would cause a tight loop or panic)");
|
||||
}
|
||||
Ok(Duration::from_secs(secs))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
// Process-global env vars: each test uses its own variable name so
|
||||
// parallel runs don't race.
|
||||
|
||||
#[test]
|
||||
fn unset_returns_default() {
|
||||
let d = duration_from_env("RADICLE_SEARCH_TEST_UNSET", 42).unwrap();
|
||||
assert_eq!(d, Duration::from_secs(42));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_returns_default() {
|
||||
let var = "RADICLE_SEARCH_TEST_EMPTY";
|
||||
unsafe { env::set_var(var, "") };
|
||||
let d = duration_from_env(var, 7);
|
||||
unsafe { env::remove_var(var) };
|
||||
assert_eq!(d.unwrap(), Duration::from_secs(7));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn valid_integer_parses() {
|
||||
let var = "RADICLE_SEARCH_TEST_VALID";
|
||||
unsafe { env::set_var(var, "15") };
|
||||
let d = duration_from_env(var, 99);
|
||||
unsafe { env::remove_var(var) };
|
||||
assert_eq!(d.unwrap(), Duration::from_secs(15));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn zero_is_rejected() {
|
||||
let var = "RADICLE_SEARCH_TEST_ZERO";
|
||||
unsafe { env::set_var(var, "0") };
|
||||
let result = duration_from_env(var, 99);
|
||||
unsafe { env::remove_var(var) };
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unparseable_is_rejected() {
|
||||
let var = "RADICLE_SEARCH_TEST_GARBAGE";
|
||||
unsafe { env::set_var(var, "not-a-number") };
|
||||
let result = duration_from_env(var, 99);
|
||||
unsafe { env::remove_var(var) };
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn negative_is_rejected() {
|
||||
let var = "RADICLE_SEARCH_TEST_NEG";
|
||||
unsafe { env::set_var(var, "-5") };
|
||||
let result = duration_from_env(var, 99);
|
||||
unsafe { env::remove_var(var) };
|
||||
assert!(result.is_err());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
pub mod repo;
|
||||
@@ -0,0 +1,290 @@
|
||||
pub mod data;
|
||||
pub use data::{Activity, Document, DocumentKey};
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use meilisearch_sdk::client::Client;
|
||||
use meilisearch_sdk::documents::{DocumentsQuery, DocumentsResults};
|
||||
use meilisearch_sdk::indexes;
|
||||
use meilisearch_sdk::settings::Settings;
|
||||
use meilisearch_sdk::task_info::TaskInfo;
|
||||
use meilisearch_sdk::tasks::Task;
|
||||
use serde::Deserialize;
|
||||
|
||||
/// Meilisearch field name for the head commit's committer timestamp.
|
||||
pub const FIELD_HEAD_COMMITTER_TIME: &str = "headCommitterTime";
|
||||
/// Meilisearch field name for the count of nodes seeding this repo.
|
||||
pub const FIELD_SEEDING_COUNT: &str = "seedingCount";
|
||||
|
||||
pub const SORTABLE: &[&str] = &[
|
||||
FIELD_HEAD_COMMITTER_TIME,
|
||||
FIELD_SEEDING_COUNT,
|
||||
Document::PRIMARY_KEY,
|
||||
];
|
||||
pub const SEARCHABLE: &[&str] = &["name", "description"];
|
||||
pub const FILTERABLE: &[&str] = &["delegates", "visibility"];
|
||||
// Promote the `sort` rule ahead of `proximity`/`attribute`/`exactness` so
|
||||
// that once a repo matches the query words (typo-tolerant), the query's
|
||||
// `seedingCount:desc` sort decides order, with finer relevance acting as the
|
||||
// tie-breaker. Meilisearch's default order keeps `sort` second-to-last, which
|
||||
// reduces seed count to a deep tie-breaker for text queries.
|
||||
pub const RANKING_RULES: &[&str] = &[
|
||||
"words",
|
||||
"typo",
|
||||
"sort",
|
||||
"proximity",
|
||||
"attribute",
|
||||
"exactness",
|
||||
];
|
||||
|
||||
/// Bounded exponential-backoff schedule shared by
|
||||
/// [`Index::configure_with_retry`] and the enqueue helpers.
|
||||
const MAX_ATTEMPTS: u32 = 10;
|
||||
const BASE_DELAY: Duration = Duration::from_secs(1);
|
||||
const MAX_DELAY: Duration = Duration::from_secs(30);
|
||||
|
||||
pub struct Index {
|
||||
pub(crate) client: Client,
|
||||
pub(crate) index: indexes::Index,
|
||||
}
|
||||
|
||||
impl Index {
|
||||
pub fn connect(url: &str, key: Option<&str>, name: &str) -> Result<Self> {
|
||||
let client = Client::new(url, key).context("failed to construct Meilisearch client")?;
|
||||
let index = client.index(name);
|
||||
Ok(Self { client, index })
|
||||
}
|
||||
|
||||
pub async fn configure(&self) -> std::result::Result<(), meilisearch_sdk::errors::Error> {
|
||||
let settings = Settings::new()
|
||||
.with_searchable_attributes(SEARCHABLE)
|
||||
.with_sortable_attributes(SORTABLE)
|
||||
.with_filterable_attributes(FILTERABLE)
|
||||
.with_ranking_rules(RANKING_RULES);
|
||||
let task = self.index.set_settings(&settings).await?;
|
||||
task.wait_for_completion(&self.client, None, None).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Run [`Self::configure`] with bounded exponential backoff. Useful at
|
||||
/// startup where Meilisearch may not be reachable yet (e.g. systemd
|
||||
/// brought both services up in parallel). Retries every error, since a
|
||||
/// failure here means the daemon can't start regardless of the cause.
|
||||
pub async fn configure_with_retry(&self) -> Result<()> {
|
||||
retry_meili("configure", |_| true, || self.configure()).await
|
||||
}
|
||||
|
||||
/// Enqueue an upsert with Meilisearch. The Meili HTTP call (which only
|
||||
/// enqueues the task) is awaited so failures to enqueue surface
|
||||
/// synchronously. Transient failures are retried with backoff. A
|
||||
/// background task then watches the task to completion and logs any
|
||||
/// async failure inside Meili.
|
||||
pub async fn upsert(&self, docs: &[Document]) -> Result<()> {
|
||||
if docs.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let task = retry_meili("upsert", is_transient, || {
|
||||
self.index.add_or_replace(docs, Some(Document::PRIMARY_KEY))
|
||||
})
|
||||
.await?;
|
||||
watch_task(self.client.clone(), task, "upsert");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Enqueue a delete with Meilisearch. Same semantics as
|
||||
/// [`Self::upsert`] — the enqueue call is awaited (retrying transient
|
||||
/// failures); task completion is watched in the background.
|
||||
pub async fn delete(&self, id: &DocumentKey) -> Result<()> {
|
||||
let task = retry_meili("delete", is_transient, || self.index.delete_document(id)).await?;
|
||||
watch_task(self.client.clone(), task, "delete");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Enqueue a batch delete with Meilisearch.
|
||||
pub async fn delete_many(&self, ids: &[DocumentKey]) -> Result<()> {
|
||||
if ids.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let task = retry_meili("delete_many", is_transient, || {
|
||||
self.index.delete_documents(ids)
|
||||
})
|
||||
.await?;
|
||||
watch_task(self.client.clone(), task, "delete_many");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// List all document ids currently present in the index. Used at
|
||||
/// bootstrap to reconcile against actual Meili state so unseeds that
|
||||
/// happened while the daemon was down don't leave orphan documents
|
||||
/// (the in-memory `seeded` cache is empty on restart, so a cache-only
|
||||
/// diff would miss them).
|
||||
pub async fn list_doc_ids(&self) -> Result<HashSet<DocumentKey>> {
|
||||
const PAGE: usize = 1000;
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct IdOnly {
|
||||
id: DocumentKey,
|
||||
}
|
||||
|
||||
let mut ids = HashSet::new();
|
||||
let mut offset = 0;
|
||||
loop {
|
||||
let mut query = DocumentsQuery::new(&self.index);
|
||||
query
|
||||
.with_limit(PAGE)
|
||||
.with_offset(offset)
|
||||
.with_fields([Document::PRIMARY_KEY]);
|
||||
let page: DocumentsResults<IdOnly> = self
|
||||
.index
|
||||
.get_documents_with(&query)
|
||||
.await
|
||||
.context("get_documents failed")?;
|
||||
let done = page.results.len() < PAGE;
|
||||
ids.extend(page.results.into_iter().map(|d| d.id));
|
||||
if done {
|
||||
break;
|
||||
}
|
||||
offset += PAGE;
|
||||
}
|
||||
Ok(ids)
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn a background task that polls `task` to completion and logs any
|
||||
/// async failure inside Meili. Fire-and-forget — the caller never awaits.
|
||||
fn watch_task(client: Client, task: TaskInfo, action: &'static str) {
|
||||
let task_uid = task.task_uid;
|
||||
tokio::spawn(async move {
|
||||
match task.wait_for_completion(&client, None, None).await {
|
||||
Ok(Task::Failed { content }) => {
|
||||
tracing::warn!("meili {action} task {task_uid} failed: {:?}", content.error);
|
||||
}
|
||||
Ok(_) => {
|
||||
tracing::debug!("meili {action} task {task_uid} completed");
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("meili {action} task {task_uid} polling failed: {e:#}");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Run a Meilisearch `op` with bounded exponential backoff. Only errors for
|
||||
/// which `retryable` returns true are retried; the rest surface immediately.
|
||||
/// `what` labels log lines. After [`MAX_ATTEMPTS`] the last error is returned
|
||||
/// with context. This keeps a momentary Meilisearch hiccup (e.g. a 408 during
|
||||
/// a rescan) from propagating up and terminating the daemon.
|
||||
async fn retry_meili<T, F, Fut, R>(what: &str, retryable: R, op: F) -> Result<T>
|
||||
where
|
||||
F: Fn() -> Fut,
|
||||
Fut: std::future::Future<Output = std::result::Result<T, meilisearch_sdk::errors::Error>>,
|
||||
R: Fn(&meilisearch_sdk::errors::Error) -> bool,
|
||||
{
|
||||
for attempt in 1..MAX_ATTEMPTS {
|
||||
match op().await {
|
||||
Ok(value) => return Ok(value),
|
||||
Err(e) if retryable(&e) => {
|
||||
let delay = BASE_DELAY
|
||||
.saturating_mul(2u32.saturating_pow(attempt - 1))
|
||||
.min(MAX_DELAY);
|
||||
tracing::warn!(
|
||||
"meili {what} attempt {attempt}/{MAX_ATTEMPTS} failed: {e}; \
|
||||
retrying in {}s",
|
||||
delay.as_secs()
|
||||
);
|
||||
tokio::time::sleep(delay).await;
|
||||
}
|
||||
Err(e) => return Err(e).context(format!("{what} failed")),
|
||||
}
|
||||
}
|
||||
op().await
|
||||
.context(format!("{what} gave up after {MAX_ATTEMPTS} attempts"))
|
||||
}
|
||||
|
||||
/// Whether a Meilisearch error is worth retrying: a momentary communication
|
||||
/// hiccup (request timeout, rate limit, or gateway error) or a
|
||||
/// transport-level connection/timeout failure. Permanent errors — malformed
|
||||
/// request, auth, parse, 4xx — are surfaced instead, since retrying them
|
||||
/// can't help.
|
||||
fn is_transient(err: &meilisearch_sdk::errors::Error) -> bool {
|
||||
use meilisearch_sdk::errors::Error;
|
||||
match err {
|
||||
// A non-JSON HTTP error from the server or a proxy in front of it:
|
||||
// 408 request timeout, 429 rate limit, 502/503/504 gateway errors.
|
||||
Error::MeilisearchCommunication(e) => {
|
||||
matches!(e.status_code, 408 | 429 | 502 | 503 | 504)
|
||||
}
|
||||
// The SDK's own timeout awaiting a task.
|
||||
Error::Timeout => true,
|
||||
// Transport-level failures: connection refused/reset, read timeout.
|
||||
Error::HttpError(e) => e.is_timeout() || e.is_connect(),
|
||||
// Malformed request, auth, parse, structured 4xx: permanent.
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
use meilisearch_sdk::errors::{Error, MeilisearchCommunicationError};
|
||||
|
||||
fn comm_err(status_code: u16) -> Error {
|
||||
Error::MeilisearchCommunication(MeilisearchCommunicationError {
|
||||
status_code,
|
||||
message: None,
|
||||
url: "http://localhost:7700".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transient_errors_are_classified() {
|
||||
for status in [408, 429, 502, 503, 504] {
|
||||
assert!(is_transient(&comm_err(status)), "{status} should retry");
|
||||
}
|
||||
for status in [400, 404] {
|
||||
assert!(!is_transient(&comm_err(status)), "{status} is permanent");
|
||||
}
|
||||
assert!(is_transient(&Error::Timeout));
|
||||
assert!(!is_transient(&Error::InvalidRequest));
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn retry_gives_up_after_max_attempts_on_transient() {
|
||||
let calls = std::cell::Cell::new(0u32);
|
||||
let result: Result<()> = retry_meili("test", is_transient, || {
|
||||
calls.set(calls.get() + 1);
|
||||
async { Err(comm_err(503)) }
|
||||
})
|
||||
.await;
|
||||
assert!(result.is_err());
|
||||
assert_eq!(calls.get(), MAX_ATTEMPTS);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn retry_surfaces_permanent_error_without_retrying() {
|
||||
let calls = std::cell::Cell::new(0u32);
|
||||
let result: Result<()> = retry_meili("test", is_transient, || {
|
||||
calls.set(calls.get() + 1);
|
||||
async { Err(comm_err(400)) }
|
||||
})
|
||||
.await;
|
||||
assert!(result.is_err());
|
||||
assert_eq!(calls.get(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn retry_recovers_after_transient_failures() {
|
||||
let calls = std::cell::Cell::new(0u32);
|
||||
let result: Result<u32> = retry_meili("test", is_transient, || {
|
||||
let n = calls.get() + 1;
|
||||
calls.set(n);
|
||||
async move { if n < 3 { Err(comm_err(503)) } else { Ok(n) } }
|
||||
})
|
||||
.await;
|
||||
assert_eq!(result.unwrap(), 3);
|
||||
assert_eq!(calls.get(), 3);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
use std::fmt;
|
||||
|
||||
use radicle::{
|
||||
git,
|
||||
identity::doc,
|
||||
prelude::{Doc, RepoId},
|
||||
};
|
||||
use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
|
||||
|
||||
/// The primary key for the index backend.
|
||||
///
|
||||
/// It is derived from a [`RepoId`], and renders to the RID without the `rad:`
|
||||
/// prefix. This is due to document IDs only allowing characters in the set
|
||||
/// `[a-zA-Z0-9_-]`.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
pub struct DocumentKey(RepoId);
|
||||
|
||||
impl fmt::Display for DocumentKey {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.write_str(&self.0.canonical())
|
||||
}
|
||||
}
|
||||
|
||||
impl DocumentKey {
|
||||
pub fn new(rid: RepoId) -> Self {
|
||||
Self(rid)
|
||||
}
|
||||
}
|
||||
|
||||
/// Serialize using the prefix-less canonical form so the value is a valid
|
||||
/// Meilisearch document id. The derived impl would forward to [`RepoId`]'s,
|
||||
/// which emits the `rad:` prefix and is rejected by Meilisearch.
|
||||
impl Serialize for DocumentKey {
|
||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
serializer.collect_str(&self.0.canonical())
|
||||
}
|
||||
}
|
||||
|
||||
impl<'de> Deserialize<'de> for DocumentKey {
|
||||
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let s = String::deserialize(deserializer)?;
|
||||
RepoId::from_canonical(&s)
|
||||
.map(Self)
|
||||
.map_err(de::Error::custom)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct Document {
|
||||
pub id: DocumentKey,
|
||||
pub rid: RepoId,
|
||||
pub name: String,
|
||||
pub description: String,
|
||||
pub default_branch: git::fmt::RefString,
|
||||
pub delegates: doc::Delegates,
|
||||
pub seeding_count: u64,
|
||||
#[serde(flatten)]
|
||||
pub activity: Activity,
|
||||
}
|
||||
|
||||
impl Document {
|
||||
pub(crate) const PRIMARY_KEY: &str = "id";
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct Activity {
|
||||
pub head: Option<git::Oid>,
|
||||
pub head_committer_time: Option<i64>,
|
||||
pub activity_timestamps: Vec<i64>,
|
||||
}
|
||||
|
||||
impl Activity {
|
||||
pub fn empty() -> Self {
|
||||
Self {
|
||||
head: None,
|
||||
head_committer_time: None,
|
||||
activity_timestamps: Vec::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Document {
|
||||
/// Construct a new [`RepoDocument`] for the given [`RepoId`], [`Doc`],
|
||||
/// [`Activity`], and seeding count.
|
||||
///
|
||||
/// If the [`Doc::visibility`] is not public or the [`Doc::project`] fails
|
||||
/// to resolve, then `None` is returned.
|
||||
pub(crate) fn new(
|
||||
rid: RepoId,
|
||||
doc: &Doc,
|
||||
activity: Activity,
|
||||
seeding_count: u64,
|
||||
) -> Option<Self> {
|
||||
if !doc.visibility().is_public() {
|
||||
return None;
|
||||
}
|
||||
|
||||
let project = match doc.project() {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::debug!("{rid}: no project payload ({e})");
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
Some(Self {
|
||||
id: DocumentKey::new(rid),
|
||||
rid,
|
||||
name: project.name().to_string(),
|
||||
description: project.description().to_string(),
|
||||
default_branch: project.default_branch().clone(),
|
||||
delegates: doc.delegates().clone(),
|
||||
seeding_count,
|
||||
activity,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,322 @@
|
||||
mod bootstrap;
|
||||
mod build;
|
||||
mod event;
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use radicle::Profile;
|
||||
use radicle::identity::RepoId;
|
||||
use radicle::node::{Event, Handle as _};
|
||||
use radicle::storage::{ReadRepository, ReadStorage};
|
||||
use tokio::sync::{RwLock, mpsc, watch};
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::index::repo;
|
||||
|
||||
const UPSERT_BATCH: usize = 100;
|
||||
const EVENT_CHANNEL_CAPACITY: usize = 1024;
|
||||
|
||||
pub struct Indexer {
|
||||
profile: Arc<Profile>,
|
||||
index: Arc<repo::Index>,
|
||||
config: Config,
|
||||
seeded: Seeded,
|
||||
}
|
||||
|
||||
/// An in-memory cache of locally-seeded repositories, given by their [`repo::RepoDocumentKey`].
|
||||
///
|
||||
/// It is used as a fast filter in the event hot path. It is refreshed at the
|
||||
/// end of every bootstrap process, and will be stale between re-scans, i.e., a
|
||||
/// new `rad seed` will not be picked up until the next re-scan.
|
||||
struct Seeded {
|
||||
inner: Arc<RwLock<HashSet<repo::DocumentKey>>>,
|
||||
}
|
||||
|
||||
impl Seeded {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
inner: Arc::new(RwLock::new(HashSet::new())),
|
||||
}
|
||||
}
|
||||
|
||||
async fn insert(&self, key: repo::DocumentKey) -> bool {
|
||||
self.inner.write().await.insert(key)
|
||||
}
|
||||
|
||||
async fn remove(&self, key: &repo::DocumentKey) -> bool {
|
||||
self.inner.write().await.remove(key)
|
||||
}
|
||||
|
||||
async fn replace(&self, keys: HashSet<repo::DocumentKey>) {
|
||||
*self.inner.write().await = keys;
|
||||
}
|
||||
|
||||
async fn contains(&self, key: &repo::DocumentKey) -> bool {
|
||||
self.inner.read().await.contains(key)
|
||||
}
|
||||
|
||||
async fn as_inner(&self) -> HashSet<repo::DocumentKey> {
|
||||
self.inner.read().await.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Indexer {
|
||||
pub fn new(profile: Arc<Profile>, index: Arc<repo::Index>, config: Config) -> Self {
|
||||
Self {
|
||||
profile,
|
||||
index,
|
||||
config,
|
||||
seeded: Seeded::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn bootstrap(&self) -> Result<()> {
|
||||
tracing::info!("starting full storage rescan");
|
||||
let profile = self.profile.clone();
|
||||
let previous_seeded = self.seeded.as_inner().await;
|
||||
|
||||
let handler = bootstrap::Bootstrap::new(previous_seeded);
|
||||
let (docs, seeded) = tokio::task::spawn_blocking(
|
||||
move || -> Result<(Vec<repo::Document>, HashSet<repo::DocumentKey>)> {
|
||||
let policies = profile.policies()?;
|
||||
let db = profile.database()?;
|
||||
let repos = profile.storage.repositories()?;
|
||||
|
||||
let seeds = repos
|
||||
.iter()
|
||||
.filter(|info| info.doc.visibility().is_public())
|
||||
.map(|info| {
|
||||
let seeding_policy = match policies.is_seeding(&info.rid) {
|
||||
Ok(true) => bootstrap::SeedingPolicy::IsSeeding,
|
||||
Ok(false) => bootstrap::SeedingPolicy::NotSeeding,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"policy lookup for {} failed: {e:#}; preserving prior state",
|
||||
info.rid
|
||||
);
|
||||
bootstrap::SeedingPolicy::LookupFailure
|
||||
}
|
||||
};
|
||||
bootstrap::RepoSeed {
|
||||
rid: info.rid,
|
||||
seeding_policy,
|
||||
}
|
||||
});
|
||||
|
||||
let plan = handler.plan(seeds);
|
||||
|
||||
let mut docs = Vec::with_capacity(plan.to_index.len());
|
||||
for info in repos
|
||||
.iter()
|
||||
.filter(|info| plan.to_index.contains(&info.rid))
|
||||
{
|
||||
match build::document(&profile, &db, info.rid, &info.doc) {
|
||||
Ok(Some(doc)) => docs.push(doc),
|
||||
Ok(None) => {}
|
||||
Err(e) => tracing::warn!("skipping {}: {e:#}", info.rid),
|
||||
}
|
||||
}
|
||||
|
||||
Ok((docs, plan.seeded))
|
||||
},
|
||||
)
|
||||
.await
|
||||
.context("bootstrap blocking task panicked")??;
|
||||
|
||||
let total = docs.len();
|
||||
tracing::info!("indexing {total} repositories");
|
||||
for chunk in docs.chunks(UPSERT_BATCH) {
|
||||
// A batch that still fails after the enqueue retries is logged and
|
||||
// skipped rather than aborting the whole rescan (which would
|
||||
// propagate to the fatal exit in main and crash-loop the daemon).
|
||||
// The next periodic rescan reconciles whatever was missed.
|
||||
if let Err(e) = self.index.upsert(chunk).await {
|
||||
tracing::warn!(
|
||||
"indexing a batch of {} repositories failed: {e:#}; \
|
||||
skipping it (next rescan will reconcile)",
|
||||
chunk.len()
|
||||
);
|
||||
}
|
||||
}
|
||||
self.seeded.replace(seeded.clone()).await;
|
||||
|
||||
let removed = match self.index.list_doc_ids().await {
|
||||
Ok(meili_ids) => {
|
||||
let orphans = bootstrap::orphans(seeded, &meili_ids);
|
||||
if !orphans.is_empty() {
|
||||
tracing::info!("removing {} orphan documents from index", orphans.len());
|
||||
if let Err(e) = self.index.delete_many(&orphans).await {
|
||||
tracing::warn!("orphan delete failed: {e:#}");
|
||||
}
|
||||
}
|
||||
orphans.len()
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"listing meili documents for reconciliation failed: {e:#}; \
|
||||
skipping orphan delete this cycle"
|
||||
);
|
||||
0
|
||||
}
|
||||
};
|
||||
tracing::info!("rescan complete ({total} indexed, {removed} removed)");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Bring the index in sync with the current local state of `rid`. If
|
||||
/// we're still seeding it, upsert a fresh document. If we've dropped
|
||||
/// it (or it's no longer indexable), delete the document and prune
|
||||
/// the seeded cache so subsequent gossip events don't re-trigger.
|
||||
pub async fn reindex(&self, rid: RepoId) -> Result<()> {
|
||||
let key = repo::DocumentKey::new(rid);
|
||||
let profile = self.profile.clone();
|
||||
let action = tokio::task::spawn_blocking(move || -> Result<ReindexAction> {
|
||||
let policies = profile.policies()?;
|
||||
// Propagate on lookup failure rather than treating it as
|
||||
// "not seeded" — a transient sqlite error would otherwise
|
||||
// delete the entry and prune the cache, hiding the rid from
|
||||
// subsequent gossip events until the next full rescan.
|
||||
if !policies.is_seeding(&rid)? {
|
||||
return Ok(ReindexAction::Delete);
|
||||
}
|
||||
let db = profile.database()?;
|
||||
let doc = {
|
||||
let repo = profile.storage.repository(rid)?;
|
||||
repo.identity_doc()?
|
||||
};
|
||||
match build::document(&profile, &db, rid, &doc)? {
|
||||
Some(doc) => Ok(ReindexAction::Upsert(Box::new(doc))),
|
||||
None => Ok(ReindexAction::Delete),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.context("reindex blocking task panicked")??;
|
||||
|
||||
match action {
|
||||
ReindexAction::Upsert(doc) => {
|
||||
self.index.upsert(std::slice::from_ref(&*doc)).await?;
|
||||
}
|
||||
ReindexAction::Delete => {
|
||||
tracing::info!("removing {rid} from index (no longer seeded)");
|
||||
self.index.delete(&key).await?;
|
||||
self.seeded.remove(&key).await;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn run(&self, mut shutdown: watch::Receiver<bool>) -> Result<()> {
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => return Ok(()),
|
||||
res = self.bootstrap() => res?,
|
||||
}
|
||||
|
||||
loop {
|
||||
let sub_shutdown = shutdown.clone();
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => return Ok(()),
|
||||
res = self.subscribe_loop(sub_shutdown) => match res {
|
||||
Ok(()) => tracing::warn!("event stream ended without error; reconnecting"),
|
||||
Err(e) => tracing::warn!("event stream error: {e:#}; reconnecting"),
|
||||
},
|
||||
}
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => return Ok(()),
|
||||
_ = tokio::time::sleep(self.config.reconnect_backoff) => {}
|
||||
}
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => return Ok(()),
|
||||
res = self.bootstrap() => {
|
||||
if let Err(e) = res {
|
||||
tracing::error!("rescan after disconnect failed: {e:#}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn subscribe_loop(&self, mut shutdown: watch::Receiver<bool>) -> Result<()> {
|
||||
let socket = self.profile.home().socket_from_env();
|
||||
tracing::info!("subscribing to node events at {}", socket.display());
|
||||
let node = radicle::Node::new(&socket);
|
||||
|
||||
let (tx, mut rx) = mpsc::channel::<Event>(EVENT_CHANNEL_CAPACITY);
|
||||
let blocking_shutdown = shutdown.clone();
|
||||
let blocking = tokio::task::spawn_blocking(move || -> Result<()> {
|
||||
let events = node
|
||||
.subscribe(Duration::from_secs(1))
|
||||
.context("subscribe to node socket failed")?;
|
||||
for event in events {
|
||||
if *blocking_shutdown.borrow() {
|
||||
break;
|
||||
}
|
||||
let event = match event {
|
||||
Ok(e) => e,
|
||||
Err(radicle::node::Error::TimedOut) => continue,
|
||||
Err(e) => {
|
||||
return Err(anyhow::Error::from(e).context("event read failed"));
|
||||
}
|
||||
};
|
||||
if tx.blocking_send(event).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
|
||||
let mut rescan_timer = tokio::time::interval(self.config.rescan_interval);
|
||||
rescan_timer.tick().await;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => break,
|
||||
event = rx.recv() => {
|
||||
let Some(event) = event else { break; };
|
||||
let Some((rid, category)) = event::classify_event(&event) else {
|
||||
tracing::debug!("ignored event: {}", event::event_kind(&event));
|
||||
continue;
|
||||
};
|
||||
let key = repo::DocumentKey::new(rid);
|
||||
let is_seeded = self.seeded.contains(&key).await;
|
||||
match event::event_action(category, is_seeded) {
|
||||
event::EventAction::Skip => {
|
||||
tracing::debug!(
|
||||
"skipping {rid} ({}): not locally seeded",
|
||||
event::event_kind(&event)
|
||||
);
|
||||
continue;
|
||||
}
|
||||
event::EventAction::Reindex => {}
|
||||
event::EventAction::DiscoverAndReindex => {
|
||||
self.seeded.insert(key).await;
|
||||
tracing::info!("discovered new local seed: {rid}");
|
||||
}
|
||||
}
|
||||
tracing::info!("reindex {rid} (event: {})", event::event_kind(&event));
|
||||
if let Err(e) = self.reindex(rid).await {
|
||||
tracing::warn!("reindex {} failed: {e:#}", rid);
|
||||
}
|
||||
}
|
||||
_ = rescan_timer.tick() => {
|
||||
if let Err(e) = self.bootstrap().await {
|
||||
tracing::warn!("periodic rescan failed: {e:#}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
blocking
|
||||
.await
|
||||
.context("event reader task panicked")?
|
||||
.context("event reader task returned an error")?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
enum ReindexAction {
|
||||
Upsert(Box<repo::Document>),
|
||||
Delete,
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
#[cfg(test)]
|
||||
mod test;
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
use radicle::identity::RepoId;
|
||||
|
||||
use crate::index::repo::DocumentKey;
|
||||
|
||||
/// One entry per public repository found in storage.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
pub(crate) struct RepoSeed {
|
||||
/// The identifier of the repository.
|
||||
pub rid: RepoId,
|
||||
/// The determined policy of the given repository.
|
||||
pub seeding_policy: SeedingPolicy,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
/// Result of determining the seeding policy of a repository.
|
||||
pub(crate) enum SeedingPolicy {
|
||||
/// The repository is being seeded.
|
||||
IsSeeding,
|
||||
/// The repository is not being seeded.
|
||||
NotSeeding,
|
||||
/// Could not determine the seeding policy of the repository.
|
||||
LookupFailure,
|
||||
}
|
||||
|
||||
/// Output of [`Bootstrap::plan`].
|
||||
#[derive(Debug, Default)]
|
||||
pub(crate) struct BootstrapPlan {
|
||||
/// Repos whose documents should be built and upserted.
|
||||
pub to_index: HashSet<RepoId>,
|
||||
/// The new seeded set, including preserved entries from prior state.
|
||||
pub seeded: HashSet<DocumentKey>,
|
||||
}
|
||||
|
||||
impl BootstrapPlan {
|
||||
fn index(&mut self, rid: RepoId) {
|
||||
self.to_index.insert(rid);
|
||||
}
|
||||
|
||||
fn seed(&mut self, key: DocumentKey) {
|
||||
self.seeded.insert(key);
|
||||
}
|
||||
}
|
||||
|
||||
/// [`Bootstrap`] handles the decision of whether incoming [`RepoSeed`]s should
|
||||
/// be indexed, and computes the new seeded set.
|
||||
pub(crate) struct Bootstrap {
|
||||
previous_seeded: HashSet<DocumentKey>,
|
||||
}
|
||||
|
||||
impl Bootstrap {
|
||||
/// Construct a new [`Bootstrap`] from a previously seeded set of [`DocumentKey`]s.
|
||||
pub fn new(previous_seeded: HashSet<DocumentKey>) -> Self {
|
||||
Self { previous_seeded }
|
||||
}
|
||||
|
||||
/// Decide which repos to index based on seeding policy results.
|
||||
///
|
||||
/// For each [`RepoSeed`]:
|
||||
/// - If it is [`SeedingPolicy::IsSeeding`] then it is marked for indexing and as
|
||||
/// part of the seeded set.
|
||||
/// - If it is [`SeedingPolicy::NotSeeding`] then it is not marked for
|
||||
/// either indexing nor the seeded set.
|
||||
/// - If it is [`SeedingPolicy::LookupFailure`] then it preserves whether it
|
||||
/// was in the previously seeded set.
|
||||
pub fn plan(self, repos: impl Iterator<Item = RepoSeed>) -> BootstrapPlan {
|
||||
repos.fold(BootstrapPlan::default(), |mut plan, repo| {
|
||||
let key = DocumentKey::new(repo.rid);
|
||||
match repo.seeding_policy {
|
||||
SeedingPolicy::IsSeeding => {
|
||||
plan.index(repo.rid);
|
||||
plan.seed(key);
|
||||
}
|
||||
SeedingPolicy::NotSeeding => {}
|
||||
SeedingPolicy::LookupFailure => {
|
||||
if self.previous_seeded.contains(&key) {
|
||||
plan.seed(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
plan
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Compute orphan document keys to delete from the index.
|
||||
///
|
||||
/// Returns keys present in `in_index` but absent from `seeded`.
|
||||
pub(crate) fn orphans(
|
||||
seeded: HashSet<DocumentKey>,
|
||||
in_index: &HashSet<DocumentKey>,
|
||||
) -> Vec<DocumentKey> {
|
||||
in_index.difference(&seeded).copied().collect()
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
mod orphans;
|
||||
mod plan;
|
||||
|
||||
use radicle::git::Oid;
|
||||
use radicle::identity::RepoId;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn rid(seed: u8) -> RepoId {
|
||||
RepoId::from(Oid::from_sha1([seed; 20]))
|
||||
}
|
||||
|
||||
fn key(seed: u8) -> DocumentKey {
|
||||
DocumentKey::new(rid(seed))
|
||||
}
|
||||
|
||||
fn seed(seed: u8, seeding_policy: SeedingPolicy) -> RepoSeed {
|
||||
RepoSeed {
|
||||
rid: rid(seed),
|
||||
seeding_policy,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
use std::collections::HashSet;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn in_sync() {
|
||||
let seeded = HashSet::from([key(1), key(2)]);
|
||||
let in_index = HashSet::from([key(1), key(2)]);
|
||||
assert!(orphans(seeded, &in_index).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn detects_removed_repo() {
|
||||
let seeded = HashSet::from([key(1)]);
|
||||
let in_index = HashSet::from([key(1), key(2)]);
|
||||
assert_eq!(orphans(seeded, &in_index), vec![key(2)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_index() {
|
||||
let seeded = HashSet::from([key(1)]);
|
||||
let in_index = HashSet::new();
|
||||
assert!(orphans(seeded, &in_index).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn all_orphaned() {
|
||||
let seeded = HashSet::new();
|
||||
let in_index = HashSet::from([key(1), key(2)]);
|
||||
let mut result = orphans(seeded, &in_index);
|
||||
result.sort_by_key(|k| format!("{k}"));
|
||||
let mut expected = vec![key(1), key(2)];
|
||||
expected.sort_by_key(|k| format!("{k}"));
|
||||
assert_eq!(result, expected);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn both_empty() {
|
||||
assert!(orphans(HashSet::new(), &HashSet::new()).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disjoint_sets() {
|
||||
let seeded = HashSet::from([key(1)]);
|
||||
let in_index = HashSet::from([key(2)]);
|
||||
assert_eq!(orphans(seeded, &in_index), vec![key(2)]);
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
use std::collections::HashSet;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn repositories(repos: impl IntoIterator<Item = RepoId>) -> HashSet<RepoId> {
|
||||
repos.into_iter().collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn seeding_new_repo() {
|
||||
let b = Bootstrap::new(HashSet::new());
|
||||
let repos = [seed(1, SeedingPolicy::IsSeeding)];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert_eq!(plan.to_index, repositories([rid(1)]));
|
||||
assert!(plan.seeded.contains(&key(1)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn seeding_previously_seeded() {
|
||||
let b = Bootstrap::new(HashSet::from([key(1)]));
|
||||
let repos = [seed(1, SeedingPolicy::IsSeeding)];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert_eq!(plan.to_index, repositories([rid(1)]));
|
||||
assert!(plan.seeded.contains(&key(1)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn not_seeding_new_repo() {
|
||||
let b = Bootstrap::new(HashSet::new());
|
||||
let repos = [seed(1, SeedingPolicy::NotSeeding)];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert!(plan.to_index.is_empty());
|
||||
assert!(!plan.seeded.contains(&key(1)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn not_seeding_drops_previously_seeded() {
|
||||
let b = Bootstrap::new(HashSet::from([key(1)]));
|
||||
let repos = [seed(1, SeedingPolicy::NotSeeding)];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert!(plan.to_index.is_empty());
|
||||
assert!(
|
||||
!plan.seeded.contains(&key(1)),
|
||||
"un-seeded repo must be dropped from seeded set"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn error_preserves_previously_seeded() {
|
||||
let b = Bootstrap::new(HashSet::from([key(1)]));
|
||||
let repos = [seed(1, SeedingPolicy::LookupFailure)];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert!(
|
||||
plan.to_index.is_empty(),
|
||||
"should not re-index on transient failure"
|
||||
);
|
||||
assert!(
|
||||
plan.seeded.contains(&key(1)),
|
||||
"cache entry must be preserved on transient failure"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn error_skips_unknown_repo() {
|
||||
let b = Bootstrap::new(HashSet::new());
|
||||
let repos = [seed(1, SeedingPolicy::LookupFailure)];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert!(plan.to_index.is_empty());
|
||||
assert!(!plan.seeded.contains(&key(1)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_input() {
|
||||
let b = Bootstrap::new(HashSet::from([key(1)]));
|
||||
let repos: [RepoSeed; 0] = [];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert!(plan.to_index.is_empty());
|
||||
assert!(plan.seeded.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mixed_policies() {
|
||||
let previous = HashSet::from([key(3)]);
|
||||
let b = Bootstrap::new(previous);
|
||||
let repos = [
|
||||
seed(1, SeedingPolicy::IsSeeding),
|
||||
seed(2, SeedingPolicy::NotSeeding),
|
||||
seed(3, SeedingPolicy::LookupFailure),
|
||||
seed(4, SeedingPolicy::IsSeeding),
|
||||
];
|
||||
let plan = b.plan(repos.into_iter());
|
||||
|
||||
assert_eq!(plan.to_index, repositories([rid(1), rid(4)]));
|
||||
assert!(plan.seeded.contains(&key(1)));
|
||||
assert!(!plan.seeded.contains(&key(2)));
|
||||
assert!(plan.seeded.contains(&key(3)));
|
||||
assert!(plan.seeded.contains(&key(4)));
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
use anyhow::{Context, Result};
|
||||
use radicle::Profile;
|
||||
use radicle::identity::RepoId;
|
||||
use radicle::node::routing::Store as _;
|
||||
use radicle::prelude::Doc;
|
||||
use radicle::storage::{ReadRepository, ReadStorage};
|
||||
use radicle_surf::Repository as SurfRepository;
|
||||
|
||||
use crate::index::repo;
|
||||
|
||||
const ONE_YEAR_SECS: i64 = 52 * 7 * 24 * 60 * 60;
|
||||
|
||||
pub(super) fn document(
|
||||
profile: &Profile,
|
||||
db: &radicle::node::Database,
|
||||
rid: RepoId,
|
||||
doc: &Doc,
|
||||
) -> Result<Option<repo::Document>> {
|
||||
let storage = &profile.storage;
|
||||
let repo = storage
|
||||
.repository(rid)
|
||||
.context(format!("opening {rid} from storage failed"))?;
|
||||
let seeding_count = db.count(&rid).unwrap_or_default() as u64;
|
||||
|
||||
let activity = match repo_activity(&repo) {
|
||||
Ok(activity) => activity,
|
||||
Err(e) => {
|
||||
tracing::debug!("{rid}: head/activity unavailable ({e:#})");
|
||||
repo::Activity::empty()
|
||||
}
|
||||
};
|
||||
|
||||
Ok(repo::Document::new(rid, doc, activity, seeding_count))
|
||||
}
|
||||
|
||||
fn repo_activity(repo: &radicle::storage::git::Repository) -> Result<repo::Activity> {
|
||||
let surf = SurfRepository::open(repo.path())?;
|
||||
let head = surf.head()?;
|
||||
let head_commit = surf.commit(head)?;
|
||||
let head_time = Some(head_commit.committer.time.seconds());
|
||||
|
||||
let cutoff = chrono::Utc::now().timestamp() - ONE_YEAR_SECS;
|
||||
let activity: Vec<i64> = surf
|
||||
.history(head)?
|
||||
.filter_map(|c| {
|
||||
let c = c.ok()?;
|
||||
let s = c.committer.time.seconds();
|
||||
if s > cutoff { Some(s) } else { None }
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(repo::Activity {
|
||||
head: Some((*head).into()),
|
||||
head_committer_time: head_time,
|
||||
activity_timestamps: activity,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
use radicle::identity::RepoId;
|
||||
use radicle::node::Event;
|
||||
|
||||
pub(super) enum EventCategory {
|
||||
/// The node successfully replicated refs for this rid; implies we seed it.
|
||||
Replication,
|
||||
/// Network gossip mentioning a rid that may or may not be locally seeded.
|
||||
Gossip,
|
||||
}
|
||||
|
||||
pub(super) fn classify_event(event: &Event) -> Option<(RepoId, EventCategory)> {
|
||||
match event {
|
||||
Event::LocalRefsAnnounced { rid, .. }
|
||||
| Event::CanonicalRefUpdated { rid, .. }
|
||||
| Event::RefsFetched { rid, .. }
|
||||
| Event::RefsSynced { rid, .. } => Some((*rid, EventCategory::Replication)),
|
||||
Event::SeedDiscovered { rid, .. }
|
||||
| Event::SeedDropped { rid, .. }
|
||||
| Event::RefsAnnounced { rid, .. } => Some((*rid, EventCategory::Gossip)),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn event_kind(event: &Event) -> &'static str {
|
||||
match event {
|
||||
Event::RefsFetched { .. } => "RefsFetched",
|
||||
Event::RefsSynced { .. } => "RefsSynced",
|
||||
Event::RefsAnnounced { .. } => "RefsAnnounced",
|
||||
Event::SeedDiscovered { .. } => "SeedDiscovered",
|
||||
Event::SeedDropped { .. } => "SeedDropped",
|
||||
Event::PeerConnected { .. } => "PeerConnected",
|
||||
Event::PeerDisconnected { .. } => "PeerDisconnected",
|
||||
Event::LocalRefsAnnounced { .. } => "LocalRefsAnnounced",
|
||||
Event::InventoryAnnounced { .. } => "InventoryAnnounced",
|
||||
Event::NodeAnnounced { .. } => "NodeAnnounced",
|
||||
Event::UploadPack(_) => "UploadPack",
|
||||
Event::CanonicalRefUpdated { .. } => "CanonicalRefUpdated",
|
||||
_ => "Unknown",
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub(super) enum EventAction {
|
||||
/// Gossip about a repo we don't seed — ignore.
|
||||
Skip,
|
||||
/// Repo we already track changed — reindex it.
|
||||
Reindex,
|
||||
/// Replication event for a repo not yet in cache — add to cache and reindex.
|
||||
DiscoverAndReindex,
|
||||
}
|
||||
|
||||
pub(super) fn event_action(category: EventCategory, is_locally_seeded: bool) -> EventAction {
|
||||
match (category, is_locally_seeded) {
|
||||
(EventCategory::Gossip, false) => EventAction::Skip,
|
||||
(EventCategory::Gossip, true) => EventAction::Reindex,
|
||||
(EventCategory::Replication, true) => EventAction::Reindex,
|
||||
(EventCategory::Replication, false) => EventAction::DiscoverAndReindex,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn gossip_not_seeded_is_skipped() {
|
||||
assert_eq!(
|
||||
event_action(EventCategory::Gossip, false),
|
||||
EventAction::Skip
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gossip_seeded_triggers_reindex() {
|
||||
assert_eq!(
|
||||
event_action(EventCategory::Gossip, true),
|
||||
EventAction::Reindex
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replication_not_seeded_discovers() {
|
||||
assert_eq!(
|
||||
event_action(EventCategory::Replication, false),
|
||||
EventAction::DiscoverAndReindex
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replication_seeded_triggers_reindex() {
|
||||
assert_eq!(
|
||||
event_action(EventCategory::Replication, true),
|
||||
EventAction::Reindex
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
pub mod config;
|
||||
pub mod index;
|
||||
pub mod indexer;
|
||||
pub mod query;
|
||||
@@ -0,0 +1,106 @@
|
||||
use std::process;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use radicle::Profile;
|
||||
use radicle_search::config::Config;
|
||||
use radicle_search::index::repo;
|
||||
use radicle_search::indexer::Indexer;
|
||||
|
||||
const HELP_MSG: &str = r#"
|
||||
Usage
|
||||
|
||||
radicle-search [<option>...]
|
||||
|
||||
Options
|
||||
|
||||
--help, -h Print help
|
||||
|
||||
Environment
|
||||
|
||||
RADICLE_SEARCH_MEILI_URL Meilisearch base URL (default: http://localhost:7700)
|
||||
RADICLE_SEARCH_MEILI_KEY Meilisearch API key (optional)
|
||||
RADICLE_SEARCH_INDEX_NAME Meilisearch index name (default: repos)
|
||||
RADICLE_SEARCH_RESCAN_SECS Periodic full rescan interval in seconds (default: 3600)
|
||||
RADICLE_SEARCH_RECONNECT_BACKOFF_SECS Reconnect delay after a dropped event stream (default: 5)
|
||||
RAD_HOME, RAD_SOCKET Standard Radicle paths
|
||||
"#;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
init_tracing();
|
||||
|
||||
if let Err(e) = parse_args() {
|
||||
eprintln!("error: {e}");
|
||||
process::exit(1);
|
||||
}
|
||||
|
||||
if let Err(e) = run().await {
|
||||
tracing::error!("fatal: {e:#}");
|
||||
process::exit(1);
|
||||
}
|
||||
}
|
||||
|
||||
fn init_tracing() {
|
||||
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("radicle_search=info,info"));
|
||||
tracing_subscriber::fmt().with_env_filter(filter).init();
|
||||
}
|
||||
|
||||
fn parse_args() -> Result<()> {
|
||||
use lexopt::prelude::*;
|
||||
|
||||
let mut parser = lexopt::Parser::from_env();
|
||||
if let Some(arg) = parser.next()? {
|
||||
match arg {
|
||||
Long("help") | Short('h') => {
|
||||
println!("{HELP_MSG}");
|
||||
process::exit(0);
|
||||
}
|
||||
_ => return Err(anyhow::anyhow!("{}", arg.unexpected())),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run() -> Result<()> {
|
||||
let config = Config::from_env()?;
|
||||
tracing::info!("meili={}, index={}", config.meili_url, config.index_name);
|
||||
|
||||
let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
|
||||
tokio::spawn(async move {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use tokio::signal::unix::{SignalKind, signal};
|
||||
let mut sigterm =
|
||||
signal(SignalKind::terminate()).expect("failed to install SIGTERM handler");
|
||||
tokio::select! {
|
||||
_ = tokio::signal::ctrl_c() => tracing::info!("SIGINT received, shutting down"),
|
||||
_ = sigterm.recv() => tracing::info!("SIGTERM received, shutting down"),
|
||||
}
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
let _ = tokio::signal::ctrl_c().await;
|
||||
tracing::info!("SIGINT received, shutting down");
|
||||
}
|
||||
let _ = shutdown_tx.send(true);
|
||||
});
|
||||
|
||||
let profile = Arc::new(Profile::load().context("failed to load Radicle profile")?);
|
||||
tracing::info!("loaded profile at {}", profile.home().path().display());
|
||||
|
||||
let index = Arc::new(repo::Index::connect(
|
||||
&config.meili_url,
|
||||
config.meili_key.as_deref(),
|
||||
&config.index_name,
|
||||
)?);
|
||||
tokio::select! {
|
||||
_ = shutdown_rx.changed() => return Ok(()),
|
||||
res = index.configure_with_retry() => res?,
|
||||
}
|
||||
|
||||
let indexer = Indexer::new(profile, index, config);
|
||||
indexer.run(shutdown_rx).await?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use meilisearch_sdk::client::Client;
|
||||
use meilisearch_sdk::indexes;
|
||||
use meilisearch_sdk::search::Selectors;
|
||||
use serde::Deserialize;
|
||||
use thiserror::Error;
|
||||
|
||||
use radicle::identity::RepoId;
|
||||
|
||||
use crate::index::repo;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum SearchError {
|
||||
#[error("search query timed out")]
|
||||
Timeout,
|
||||
#[error("meilisearch error: {0}")]
|
||||
Meili(#[from] meilisearch_sdk::errors::Error),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub enum SortField {
|
||||
HeadCommitterTime,
|
||||
SeedingCount,
|
||||
}
|
||||
|
||||
impl SortField {
|
||||
fn field(self) -> &'static str {
|
||||
match self {
|
||||
Self::HeadCommitterTime => repo::FIELD_HEAD_COMMITTER_TIME,
|
||||
Self::SeedingCount => repo::FIELD_SEEDING_COUNT,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A read-only Meilisearch client for obtaining pre-sorted repo listings
|
||||
/// and full-text search results. Document construction lives in the
|
||||
/// indexer; this client only reads.
|
||||
#[derive(Clone)]
|
||||
pub struct SearchClient {
|
||||
index: indexes::Index,
|
||||
query_timeout: Duration,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RidHit {
|
||||
rid: String,
|
||||
}
|
||||
|
||||
impl SearchClient {
|
||||
pub fn new(
|
||||
url: &str,
|
||||
api_key: Option<&str>,
|
||||
index_name: &str,
|
||||
query_timeout: Duration,
|
||||
) -> Result<Self, SearchError> {
|
||||
let client = Client::new(url, api_key)?;
|
||||
let index = client.index(index_name);
|
||||
Ok(Self {
|
||||
index,
|
||||
query_timeout,
|
||||
})
|
||||
}
|
||||
|
||||
/// Return a sorted, paginated list of RIDs. Always sorts descending —
|
||||
/// the only useful direction for activity/seeding.
|
||||
pub async fn sorted_rids(
|
||||
&self,
|
||||
sort: SortField,
|
||||
offset: usize,
|
||||
limit: usize,
|
||||
) -> Result<Vec<RepoId>, SearchError> {
|
||||
let sort_clause = format!("{}:desc", sort.field());
|
||||
let sort_attrs = [sort_clause.as_str()];
|
||||
let attrs = ["rid"];
|
||||
|
||||
let mut query = self.index.search();
|
||||
query
|
||||
.with_sort(&sort_attrs)
|
||||
.with_offset(offset)
|
||||
.with_limit(limit)
|
||||
.with_attributes_to_retrieve(Selectors::Some(&attrs));
|
||||
|
||||
let result = tokio::time::timeout(self.query_timeout, query.execute::<RidHit>())
|
||||
.await
|
||||
.map_err(|_| SearchError::Timeout)??;
|
||||
|
||||
let rids = result
|
||||
.hits
|
||||
.into_iter()
|
||||
.filter_map(|hit| hit.result.rid.parse::<RepoId>().ok())
|
||||
.collect();
|
||||
Ok(rids)
|
||||
}
|
||||
|
||||
/// Free-text search by name / description. Meilisearch ranks by its
|
||||
/// built-in relevance rules (typo-tolerant prefix match); ties — and
|
||||
/// the empty-query case — fall back to seedingCount desc.
|
||||
pub async fn search_by_query(
|
||||
&self,
|
||||
query: &str,
|
||||
offset: usize,
|
||||
limit: usize,
|
||||
) -> Result<Vec<RepoId>, SearchError> {
|
||||
let sort_clause = format!("{}:desc", repo::FIELD_SEEDING_COUNT);
|
||||
let sort_attrs = [sort_clause.as_str()];
|
||||
let attrs = ["rid"];
|
||||
|
||||
let mut search = self.index.search();
|
||||
search
|
||||
.with_query(query)
|
||||
.with_sort(&sort_attrs)
|
||||
.with_offset(offset)
|
||||
.with_limit(limit)
|
||||
.with_attributes_to_retrieve(Selectors::Some(&attrs));
|
||||
|
||||
let result = tokio::time::timeout(self.query_timeout, search.execute::<RidHit>())
|
||||
.await
|
||||
.map_err(|_| SearchError::Timeout)??;
|
||||
|
||||
let rids = result
|
||||
.hits
|
||||
.into_iter()
|
||||
.filter_map(|hit| hit.result.rid.parse::<RepoId>().ok())
|
||||
.collect();
|
||||
Ok(rids)
|
||||
}
|
||||
}
|
||||
@@ -31,11 +31,11 @@ services:
|
||||
volumes:
|
||||
- .:/work
|
||||
- e2e-node-modules:/work/node_modules
|
||||
# Container-private cargo crate cache and httpd build dir, so the
|
||||
# Container-private cargo crate cache and workspace build dir, so the
|
||||
# Rust build is incremental across runs and never mixes with the
|
||||
# host's (different glibc/rustc) target artifacts.
|
||||
- e2e-cargo-registry:/usr/local/cargo/registry
|
||||
- e2e-httpd-target:/work/radicle-httpd/target
|
||||
- e2e-httpd-target:/work/target
|
||||
|
||||
volumes:
|
||||
e2e-node-modules:
|
||||
|
||||
@@ -62,11 +62,11 @@
|
||||
};
|
||||
inherit (pkgs) lib;
|
||||
|
||||
rustToolchain = pkgs.rust-bin.fromRustupToolchainFile ./radicle-httpd/rust-toolchain;
|
||||
rustToolchain = pkgs.rust-bin.fromRustupToolchainFile ./rust-toolchain;
|
||||
craneLib = (crane.mkLib pkgs).overrideToolchain rustToolchain;
|
||||
basicArgs = {
|
||||
pname = "radicle-httpd";
|
||||
src = ./radicle-httpd;
|
||||
pname = "radicle-explorer-rust";
|
||||
src = craneLib.cleanCargoSource ./.;
|
||||
strictDeps = true;
|
||||
};
|
||||
in {
|
||||
@@ -75,6 +75,7 @@
|
||||
checks = {
|
||||
radicle-explorer = self.packages.${system}.radicle-explorer.override {doCheck = true;};
|
||||
radicle-httpd = self.packages.${system}.radicle-httpd.overrideAttrs (_: {doCheck = true;});
|
||||
radicle-search = self.packages.${system}.radicle-search.overrideAttrs (_: {doCheck = true;});
|
||||
};
|
||||
|
||||
packages = {
|
||||
@@ -132,7 +133,7 @@
|
||||
|
||||
radicle-httpd = craneLib.buildPackage (basicArgs
|
||||
// {
|
||||
inherit (craneLib.crateNameFromCargoToml {cargoToml = ./radicle-httpd + "/Cargo.toml";}) pname version;
|
||||
inherit (craneLib.crateNameFromCargoToml {cargoToml = ./crates/radicle-httpd + "/Cargo.toml";}) pname version;
|
||||
cargoArtifacts = craneLib.buildDepsOnly basicArgs;
|
||||
|
||||
nativeBuildInputs = with pkgs;
|
||||
@@ -167,6 +168,24 @@
|
||||
done
|
||||
'';
|
||||
});
|
||||
|
||||
radicle-search = craneLib.buildPackage (basicArgs
|
||||
// {
|
||||
inherit (craneLib.crateNameFromCargoToml {cargoToml = ./crates/radicle-search + "/Cargo.toml";}) pname version;
|
||||
cargoArtifacts = craneLib.buildDepsOnly basicArgs;
|
||||
|
||||
nativeBuildInputs = with pkgs;
|
||||
[
|
||||
git
|
||||
]
|
||||
++ lib.optionals pkgs.stdenv.isDarwin (with pkgs; [
|
||||
libiconv
|
||||
darwin.apple_sdk.frameworks.Security
|
||||
]);
|
||||
|
||||
cargoExtraArgs = "-p radicle-search";
|
||||
doCheck = false;
|
||||
});
|
||||
};
|
||||
|
||||
devShells.default = pkgs.mkShell {
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
1.87.0
|
||||
@@ -1,113 +0,0 @@
|
||||
use std::num::NonZeroUsize;
|
||||
use std::{collections::HashMap, process};
|
||||
|
||||
use axum_listener::DualAddr;
|
||||
use radicle::prelude::RepoId;
|
||||
use radicle::version::Version;
|
||||
use radicle_httpd as httpd;
|
||||
|
||||
pub const VERSION: Version = Version {
|
||||
name: "radicle-httpd",
|
||||
commit: env!("GIT_HEAD"),
|
||||
version: env!("RADICLE_VERSION"),
|
||||
timestamp: env!("SOURCE_DATE_EPOCH"),
|
||||
};
|
||||
|
||||
pub const HELP_MSG: &str = r#"
|
||||
Usage
|
||||
|
||||
radicle-httpd [<option>...]
|
||||
|
||||
Options
|
||||
|
||||
--listen <address> Address to listen on: TCP address (e.g., 127.0.0.1:8080)
|
||||
or Unix socket path (e.g., /tmp/radicle.sock)
|
||||
(default: 0.0.0.0:8080)
|
||||
--alias, -a <alias> <rid> Provide alias and RID pairs to shorten git clone commands for repositories,
|
||||
e.g. heartwood and rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5 to produce https://seed.radicle.dev/heartwood.git
|
||||
--cache <number> Max amount of items in cache for /tree endpoints (default: 100)
|
||||
--version, -v Print program version
|
||||
--help, -h Print help
|
||||
"#;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
// SAFETY: The logger is only initialized once.
|
||||
httpd::logger::init().unwrap();
|
||||
tracing::info!("starting http daemon..");
|
||||
tracing::info!("version {} ({})", env!("RADICLE_VERSION"), env!("GIT_HEAD"));
|
||||
|
||||
let options = parse_options()?;
|
||||
|
||||
match httpd::run(options).await {
|
||||
Ok(()) => {}
|
||||
Err(err) => {
|
||||
tracing::error!("Fatal: {:#}", err);
|
||||
process::exit(1);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Parse command-line arguments into HTTP options.
|
||||
fn parse_options() -> Result<httpd::Options, lexopt::Error> {
|
||||
use lexopt::prelude::*;
|
||||
|
||||
let mut parser = lexopt::Parser::from_env();
|
||||
let mut listen = None;
|
||||
let mut aliases = HashMap::new();
|
||||
let mut cache = Some(httpd::DEFAULT_CACHE_SIZE);
|
||||
|
||||
while let Some(arg) = parser.next()? {
|
||||
match arg {
|
||||
Long("listen") => {
|
||||
let addr: DualAddr = parser.value()?.parse()?;
|
||||
|
||||
#[cfg(unix)]
|
||||
// Get socket path and remove it if existing
|
||||
if let DualAddr::Uds(socket_path) = &addr {
|
||||
if let Some(path) = socket_path.as_pathname() {
|
||||
if path.exists() {
|
||||
tracing::info!("Removing existing socket path at {}", path.display());
|
||||
if let Err(e) = std::fs::remove_file(path) {
|
||||
tracing::error!("{e}");
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tracing::error!("Provided socket address isn't a valid path.");
|
||||
process::exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
listen = Some(addr);
|
||||
}
|
||||
Long("alias") | Short('a') => {
|
||||
let alias: String = parser.value()?.parse()?;
|
||||
let id: RepoId = parser.value()?.parse()?;
|
||||
|
||||
aliases.insert(alias, id);
|
||||
}
|
||||
Long("version") | Short('v') => {
|
||||
if let Err(e) = VERSION.write(std::io::stdout()) {
|
||||
tracing::error!("{e}");
|
||||
process::exit(1);
|
||||
};
|
||||
process::exit(0);
|
||||
}
|
||||
Long("cache") => {
|
||||
let size = parser.value()?.parse()?;
|
||||
cache = NonZeroUsize::new(size);
|
||||
}
|
||||
Long("help") | Short('h') => {
|
||||
println!("{HELP_MSG}");
|
||||
process::exit(0);
|
||||
}
|
||||
_ => return Err(arg.unexpected()),
|
||||
}
|
||||
}
|
||||
Ok(httpd::Options {
|
||||
aliases,
|
||||
listen: listen.unwrap_or_else(|| DualAddr::Tcp(([0, 0, 0, 0], 8080).into())),
|
||||
cache,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
1.95.0
|
||||
@@ -3,19 +3,13 @@ set -e
|
||||
|
||||
echo "🔨 Compiling local radicle-httpd..."
|
||||
|
||||
# Ensure radicle-httpd directory exists
|
||||
if [ ! -d "radicle-httpd" ]; then
|
||||
echo "❌ radicle-httpd directory not found"
|
||||
exit 1
|
||||
fi
|
||||
# radicle-httpd lives in the Cargo workspace under crates/. Building the
|
||||
# workspace member places the binary at the workspace-root target/, not under
|
||||
# the crate directory.
|
||||
cargo build -p radicle-httpd
|
||||
|
||||
# Compile radicle-httpd (debug build for faster compilation)
|
||||
cd radicle-httpd
|
||||
cargo build
|
||||
cd ..
|
||||
|
||||
# Verify binary was created
|
||||
if [ ! -f "radicle-httpd/target/debug/radicle-httpd" ]; then
|
||||
BIN="target/debug/radicle-httpd"
|
||||
if [ ! -f "$BIN" ]; then
|
||||
echo "❌ Compilation failed - binary not found"
|
||||
exit 1
|
||||
fi
|
||||
@@ -24,6 +18,6 @@ fi
|
||||
mkdir -p tests/tmp/bin/httpd/local
|
||||
|
||||
# Copy binary to test location
|
||||
cp radicle-httpd/target/debug/radicle-httpd tests/tmp/bin/httpd/local/radicle-httpd
|
||||
cp "$BIN" tests/tmp/bin/httpd/local/radicle-httpd
|
||||
|
||||
echo "✅ Local radicle-httpd compiled and ready at tests/tmp/bin/httpd/local/radicle-httpd"
|
||||
|
||||
@@ -13,7 +13,7 @@ cd /work
|
||||
# The container runs as root, so anything it writes into the bind-mounted repo
|
||||
# would otherwise be root-owned on the host. Hand those paths back to the
|
||||
# invoking user (HOST_UID/HOST_GID set by scripts/test-e2e-docker) on exit.
|
||||
# radicle-httpd/target lives in a named volume, so it's not host-visible.
|
||||
# The workspace target/ lives in a named volume, so it's not host-visible.
|
||||
cleanup() {
|
||||
if [ -n "${HOST_UID:-}" ]; then
|
||||
chown -R "$HOST_UID:${HOST_GID:-$HOST_UID}" \
|
||||
|
||||
Reference in New Issue
Block a user