Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 1 addition & 2 deletions crates/pm/src/cmd/view.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,9 @@ pub async fn view(package_spec: &str) -> Result<()> {
tracing::debug!("Fetched package info: {full_manifest:?}");

// Resolve version and get full VersionManifest (with all display fields)
let version_list: Vec<String> = full_manifest.versions.clone();
let resolved_version = utoo_ruborist::resolver::version::resolve_target_version(
&full_manifest.dist_tags,
&version_list,
&full_manifest.versions,
version_spec,
)
.map_err(|e| {
Expand Down
6 changes: 5 additions & 1 deletion crates/ruborist/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,11 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-

# Native-only dependencies (not compiled for WASM)
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
libc = "0.2"
libc = "0.2"
# Dedicated CPU thread pool for simd_json manifest parsing — keeps the
# tokio runtime free of CPU-bound work so in-flight network IO is not
# blocked by sibling parses. WASM falls back to inline parse.
rayon = "1.10"

# Browser handles TLS via the fetch API, so reqwest needs no rustls
# feature on this target.
Expand Down
27 changes: 27 additions & 0 deletions crates/ruborist/src/model/manifest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,33 @@ impl FullManifest {
}
}

/// Re-parse the raw manifest bytes on rayon's CPU thread pool to extract
/// a single version. Each resolved package triggers one full re-parse via
/// `simd_json::to_borrowed_value`, which on hot paths (300+ resolves per
/// install) accumulates into the dominant CPU bucket.
///
/// Returns `Arc<CoreVersionManifest>` so callers (e.g. UnifiedRegistry's
/// version-manifest cache) can store and clone the result without
/// deep-copying the underlying `HashMap`s. Wasm32 falls back to inline
/// parse — no rayon, no thread pool.
pub async fn extract_core_version_off_runtime(
full: std::sync::Arc<FullManifest>,
version: String,
) -> Option<std::sync::Arc<CoreVersionManifest>> {
#[cfg(not(target_arch = "wasm32"))]
{
let (tx, rx) = tokio::sync::oneshot::channel();
rayon::spawn(move || {
let _ = tx.send(full.get_core_version(&version).map(std::sync::Arc::new));
});
rx.await.ok().flatten()

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The current implementation with .ok().flatten() silently swallows panics that might occur within the rayon::spawn task (e.g., in full.get_core_version). This can lead to confusing behavior, as a panic during parsing would manifest as a "version not found" error downstream, which is misleading.

According to the repository's general rules, panics should be treated as unrecoverable bugs and not be silently recovered from. To adhere to this, you should propagate the panic to the calling tokio task.

You can achieve this by replacing .ok().flatten() with .expect(...) to ensure that a panic in the spawned task will cause the awaiting task to panic as well.

Suggested change
rx.await.ok().flatten()
rx.await.expect("rayon task for extract_core_version_off_runtime panicked")
References
  1. Do not implement recovery logic for panics. Panics should be treated as unrecoverable bugs that need to be fixed, not as transient, recoverable errors.

}
#[cfg(target_arch = "wasm32")]
{
full.get_core_version(&version).map(std::sync::Arc::new)
}
}

/// Deserialize a versions map by extracting only the keys, skipping all values.
///
/// Uses `IgnoredAny` to skip over version manifest JSON objects without allocating,
Expand Down
2 changes: 1 addition & 1 deletion crates/ruborist/src/resolver/preload.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ use crate::traits::progress::{BuildEvent, EventReceiver};
use crate::traits::registry::RegistryClient;

/// Default concurrency limit for manifest fetching
pub const DEFAULT_CONCURRENCY: usize = 64;
pub const DEFAULT_CONCURRENCY: usize = 256;

/// A dependency spec: (name, version_spec)
pub type Dep = (String, String);
Expand Down
6 changes: 2 additions & 4 deletions crates/ruborist/src/resolver/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -137,14 +137,12 @@ pub fn resolve_from_manifest<E: std::error::Error + 'static>(
manifest: &FullManifest,
spec: &str,
) -> Result<ResolvedPackage, ResolveError<E>> {
let version_list: Vec<String> = manifest.versions.clone();

if version_list.is_empty() {
if manifest.versions.is_empty() {
return Err(ResolveError::NoVersions(manifest.name.clone()));
}

// Resolve version using shared logic
let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec)
let resolved_version = resolve_target_version(&manifest.dist_tags, &manifest.versions, spec)
.map_err(|e| ResolveError::Version(format!("{}@{}: {}", manifest.name, spec, e)))?;

// Get manifest for resolved version (lazy: parse from Value on demand)
Expand Down
48 changes: 40 additions & 8 deletions crates/ruborist/src/service/manifest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,35 @@ use super::fetch::{
use super::http::get_client;
use crate::model::manifest::{CoreVersionManifest, FullManifest};

/// Parse JSON bytes on rayon's CPU thread pool (native) or inline
/// (wasm32). Keeps the tokio runtime free of `simd_json` CPU work so
/// other in-flight manifest fetches can keep driving network IO while
/// this one is parsing. Rayon's global pool work-steals across
/// `num_cpus` threads, so concurrent manifest parses parallelise.
async fn parse_json_off_runtime<T>(bytes: Vec<u8>) -> Result<T, anyhow::Error>
where
T: serde::de::DeserializeOwned + Send + 'static,
{
#[cfg(not(target_arch = "wasm32"))]
{
let (tx, rx) = tokio::sync::oneshot::channel();
rayon::spawn(move || {
let mut buf = bytes;
let result = simd_json::serde::from_slice::<T>(&mut buf)
.map_err(|e| anyhow!("JSON parse error: {e}"));
// Receiver dropped means the awaiting future was cancelled.
let _ = tx.send(result);
});
rx.await
.map_err(|e| anyhow!("rayon parse channel closed: {e}"))?
Comment on lines +35 to +36

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

If a panic occurs within the rayon::spawn task, the current implementation treats it as a recoverable error by mapping it to an anyhow! result. According to the repository rules, panics should be treated as unrecoverable bugs. To adhere to this, you should propagate the panic using .expect() instead of converting it into a Result error.

Suggested change
rx.await
.map_err(|e| anyhow!("rayon parse channel closed: {e}"))?
rx.await.expect("rayon task panicked during JSON parsing")
References
  1. Do not implement recovery logic for panics. Panics should be treated as unrecoverable bugs that need to be fixed, not as transient, recoverable errors.

}
#[cfg(target_arch = "wasm32")]
{
let mut buf = bytes;
simd_json::serde::from_slice::<T>(&mut buf).map_err(|e| anyhow!("JSON parse error: {e}"))
}
}

/// Result of a full manifest fetch with ETag support.
/// Transient return value, immediately destructured — Box not needed.
#[allow(clippy::large_enum_variant)]
Expand Down Expand Up @@ -91,11 +120,13 @@ pub async fn fetch_full_manifest(opts: FetchManifestOptions<'_>) -> Result<Fetch
.await
.map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))?
.to_vec();
// Save raw bytes before simd_json mutates the parse buffer.
let mut parse_buf = raw_bytes.clone();
let mut manifest: FullManifest =
simd_json::serde::from_slice(&mut parse_buf)
.map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}")))?;
// simd_json mutates the parse buffer; clone so the raw
// bytes survive for `manifest.raw`. Parse runs off-runtime
// on rayon (native) so it cannot block sibling fetches.
let parse_buf = raw_bytes.clone();
let mut manifest: FullManifest = parse_json_off_runtime(parse_buf)
.await
.map_err(FetchError::Permanent)?;
manifest.raw = std::sync::Arc::from(raw_bytes);

Ok(FetchManifestResult::Ok(manifest, new_etag))
Expand Down Expand Up @@ -182,13 +213,14 @@ pub async fn fetch_version_manifest(
.map_err(classify_reqwest_error)?;

if response.status().is_success() {
let mut bytes = response
let bytes = response
.bytes()
.await
.map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))?
.to_vec();
simd_json::serde::from_slice(&mut bytes)
.map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}")))
parse_json_off_runtime::<CoreVersionManifest>(bytes)
.await
.map_err(FetchError::Permanent)
} else {
Err(classify_status(response.status(), &url))
}
Expand Down
62 changes: 47 additions & 15 deletions crates/ruborist/src/service/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -244,11 +244,19 @@ impl UnifiedRegistry {
.await
.map_err(RegistryError)?
{
manifest::FetchManifestResult::Ok(manifest, new_etag) => {
manifest::FetchManifestResult::Ok(mut manifest, new_etag) => {
// Move version_list / dist_tags out of `manifest`
// into `VersionsInfo` instead of cloning. The cached
// `Arc<FullManifest>` keeps `raw` + scalar metadata;
// the canonical version_list / dist_tags now live in
// the `versions_info` cache slot only. Readers that
// need them go through `cache.get_versions(name)`.
let version_list = std::mem::take(&mut manifest.versions);
let dist_tags = std::mem::take(&mut manifest.dist_tags);
let versions_info = Arc::new(VersionsInfo {
versions: Versions {
version_list: manifest.versions.clone(),
dist_tags: manifest.dist_tags.clone(),
version_list,
dist_tags,
},
etag: new_etag,
last_updated: current_timestamp_secs(),
Expand All @@ -269,18 +277,20 @@ impl UnifiedRegistry {
self.cache.set_versions(name.to_string(), versions_info);
} else {
// Persistent store corrupted/missing, fetch fresh (without etag).
let (manifest, new_etag) = manifest::fetch_full_manifest_fresh(
let (mut manifest, new_etag) = manifest::fetch_full_manifest_fresh(
&self.registry_url,
name,
manifest::MetadataFormat::Abbreviated,
)
.await
.map_err(RegistryError)?;

let version_list = std::mem::take(&mut manifest.versions);
let dist_tags = std::mem::take(&mut manifest.dist_tags);
let versions_info = Arc::new(VersionsInfo {
versions: Versions {
version_list: manifest.versions.clone(),
dist_tags: manifest.dist_tags.clone(),
version_list,
dist_tags,
},
etag: new_etag,
last_updated: current_timestamp_secs(),
Expand Down Expand Up @@ -362,9 +372,8 @@ impl UnifiedRegistry {
// version from the full manifest. `resolve_full_manifest`
// is itself inflight-gated, so concurrent specs for the
// same name share one full-manifest fetch.
let (resolved_version, manifest) =
let (resolved_version, arc) =
self.resolve_via_full_manifest(name, spec).await?;
let arc = Arc::new(manifest);
self.cache.set_version_manifest(
name.to_string(),
spec.to_string(),
Expand Down Expand Up @@ -426,16 +435,39 @@ impl UnifiedRegistry {
&self,
name: &str,
spec: &str,
) -> Result<(String, CoreVersionManifest), RegistryError> {
) -> Result<(String, Arc<CoreVersionManifest>), RegistryError> {
match self.resolve_full_manifest(name).await? {
FullManifestResult::Full(full) => {
if full.versions.is_empty() {
// Canonical `version_list` / `dist_tags` live in the
// `versions_info` cache slot; the parallel slot is filled by
// the same code path that filled `full_manifests`. We don't
// duplicate them on the cached `Arc<FullManifest>`.
let versions_info = self.cache.get_versions(name).ok_or_else(|| {
RegistryError(anyhow!("Versions cache not found for {}", name))
})?;
if versions_info.versions.version_list.is_empty() {
return Err(RegistryError(anyhow!("No versions available for {}", name)));
}
let resolved_version =
resolve_target_version(&full.dist_tags, &full.versions, spec)
.map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?;
let core = full.get_core_version(&resolved_version).ok_or_else(|| {
let resolved_version = resolve_target_version(
&versions_info.versions.dist_tags,
&versions_info.versions.version_list,
spec,
)
.map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?;
// Short-circuit: a sibling spec may have already resolved to
// this same version and warmed the cache. Skip the rayon
// reparse (the dominant cost on the resolve hot path) and
// share the existing `Arc<CoreVersionManifest>` instead of
// deep-cloning its HashMaps.
if let Some(cached) = self.cache.get_version_manifest(name, &resolved_version) {
return Ok((resolved_version, cached));
}
let core = crate::model::manifest::extract_core_version_off_runtime(
Arc::clone(&full),
resolved_version.clone(),
)
.await
.ok_or_else(|| {
RegistryError(anyhow!(
"Version {} not found in manifest for {}",
resolved_version,
Expand Down Expand Up @@ -469,7 +501,7 @@ impl UnifiedRegistry {
})
.await
.map_err(RegistryError)?;
Ok((resolved_version, manifest))
Ok((resolved_version, Arc::new(manifest)))
}
}
}
Expand Down
18 changes: 10 additions & 8 deletions crates/ruborist/src/traits/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -157,11 +157,11 @@ pub trait RegistryClient {
) -> impl Future<Output = Result<Arc<CoreVersionManifest>, Self::Error>> {
async move {
let manifest = self.fetch_full_manifest(name).await?;
let version_list: Vec<String> = manifest.versions.clone();

// Resolve version using shared logic
let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec)
.map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?;
let resolved_version =
resolve_target_version(&manifest.dist_tags, &manifest.versions, spec)
.map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?;

manifest
.get_core_version(&resolved_version)
Expand Down Expand Up @@ -227,19 +227,21 @@ pub trait RegistryClient {
fetch_spec
);
let full_manifest = self.fetch_full_manifest(&fetch_name).await?;
let version_list: Vec<String> = full_manifest.versions.clone();

if version_list.is_empty() {
if full_manifest.versions.is_empty() {
return Err(RegistryError(anyhow::anyhow!(
"No versions available for {}",
fetch_name
))
.into());
}

let resolved_version =
resolve_target_version(&full_manifest.dist_tags, &version_list, &fetch_spec)
.map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?;
let resolved_version = resolve_target_version(
&full_manifest.dist_tags,
&full_manifest.versions,
&fetch_spec,
)
.map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?;

let version_manifest = full_manifest
.get_core_version(&resolved_version)
Expand Down
Loading