mirror/pnpm
1
0
Fork 0
mirror of https://github.com/pnpm/pnpm.git synced 2026-07-17 11:12:25 -04:00
Code Issues Packages Projects Releases Wiki Activity

Revert "perf(pnpr): stream package frames from the frozen-lockfile short-circuit"

Browse Source 
This reverts commit 7047c2f33a.
This commit is contained in:
Zoltan Kochan
2026-06-07 21:29:13 +02:00
parent 7047c2f33a
commit f0dfb0da1d
2 changed files with 4 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
pacquet/crates/pnpr-client/tests/integration.rs
View File

@@ -233,36 +233,6 @@ async fn streams_resolved_packages_before_the_lockfile() {
}
}
#[tokio::test]
async fn streams_packages_when_reusing_a_frozen_lockfile() {
let registry = TestRegistry::start();
let (pnpr_url, _storage) = start_pnpr().await;
let client = PnprClient::new(pnpr_url);
let first = client
.resolve(options(&registry.url(), deps([("@foo/no-deps", "1.0.0")])))
.await
.expect("first install");
let mut opts = options(&registry.url(), deps([("@foo/no-deps", "1.0.0")]));
opts.lockfile = Some(first.lockfile.clone());
opts.frozen_lockfile = true;
let mut streamed: Vec<String> = Vec::new();
let second = client
.resolve_streaming(opts, |pkg| {
assert!(!pkg.integrity.is_empty(), "a package frame carries an integrity");
assert!(pkg.tarball.starts_with("http"), "a package frame carries a tarball URL");
streamed.push(pkg.id);
})
.await
.expect("frozen lockfile reuse should succeed");
assert!(!streamed.is_empty(), "frozen reuse should stream package frames before `done`");
assert_eq!(second.lockfile, first.lockfile, "frozen reuse should return the input lockfile");
}
/// Optional dependencies must reach the server in the request, not be
/// silently dropped, so the resolved lockfile includes their edges.
#[tokio::test]

73
pnpr/crates/pnpr/src/resolver.rs
View File

@@ -53,10 +53,9 @@ use axum::{
http::{StatusCode, header},
response::Response,
};
use futures_util::StreamExt as _;
use indexmap::IndexMap;
use pacquet_config::Config as PacquetConfig;
use pacquet_lockfile::{Lockfile, LockfileResolution, PackageKey, PackageMetadata};
use pacquet_lockfile::Lockfile;
use pacquet_lockfile_verification::{collect_resolution_policy_violations, hash_lockfile};
use pacquet_network::{AuthHeaders, ThrottledClient};
use pacquet_package_manager::build_resolution_verifiers;
@@ -239,11 +238,9 @@ pub(crate) async fn handle_resolve(runtime: &Resolver, body: Bytes) -> Response
}
// Short-circuit paths that produce the whole lockfile without an
// incremental tree walk. Still stream package frames first so the
// client can prefetch tarballs while the terminal lockfile frame is
// serialized and transferred.
// incremental tree walk: nothing to stream, so emit only `done`.
if let Some(lockfile) = resolve::fresh_frozen_input_lockfile(config, &request) {
return ndjson_lockfile_stream_response(config, lockfile);
return ndjson_single_frame(&done_frame(&lockfile));
}
let resolution_cache_key = if request.auth_headers.is_empty() && request.lockfile.is_none() {
resolution_cache_key(config, &request)
@@ -254,7 +251,7 @@ pub(crate) async fn handle_resolve(runtime: &Resolver, body: Bytes) -> Response
&& let Some(lockfile) =
cached_resolution(&runtime.resolution_cache, runtime.resolution_cache_ttl, key)
{
return ndjson_lockfile_stream_response(config, lockfile);
return ndjson_single_frame(&done_frame(&lockfile));
}
// Streaming resolve. Run it in a detached task that pushes one
@@ -444,68 +441,6 @@ fn ndjson_single_frame(frame: &[u8]) -> Response {
.expect("binary response is always valid")
}
/// A 200 NDJSON response for a lockfile that is already known before
/// the response starts. Registry package frames come first to unlock
/// client-side tarball prefetch, then the terminal `done` frame carries
/// the full lockfile exactly as before.
fn ndjson_lockfile_stream_response(config: &'static PacquetConfig, lockfile: Lockfile) -> Response {
let package_frames = package_frames_for_lockfile(config, &lockfile);
let stream = futures_util::stream::iter(
package_frames.into_iter().map(|frame| Ok::<_, std::io::Error>(Bytes::from(frame))),
)
.chain(futures_util::stream::once(async move {
Ok::<_, std::io::Error>(Bytes::from(done_frame(&lockfile)))
}));
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, NDJSON_CONTENT_TYPE)
.body(Body::from_stream(stream))
.expect("streaming lockfile response is always valid")
}
fn package_frames_for_lockfile(config: &PacquetConfig, lockfile: &Lockfile) -> Vec<Vec<u8>> {
let Some(packages) = lockfile.packages.as_ref() else { return Vec::new() };
let mut packages: Vec<_> = packages.iter().collect();
packages.sort_by_key(|(key, _)| key.to_string());
packages
.into_iter()
.filter_map(|(package_key, metadata)| {
package_frame_for_lockfile_entry(config, package_key, metadata)
})
.collect()
}
fn package_frame_for_lockfile_entry(
config: &PacquetConfig,
package_key: &PackageKey,
metadata: &PackageMetadata,
) -> Option<Vec<u8>> {
let LockfileResolution::Registry(resolution) = &metadata.resolution else {
return None;
};
let package_key = package_key.without_peer();
let name = package_key.name.to_string();
let version = package_key.suffix.to_string();
let frame = serde_json::json!({
"type": "package",
"id": package_key.to_string(),
"name": name,
"version": version,
"integrity": resolution.integrity.to_string(),
"tarball": registry_tarball_url(&config.registry, &package_key),
});
ndjson_line(&frame).ok()
}
fn registry_tarball_url(registry: &str, package_key: &PackageKey) -> String {
let registry = registry.strip_suffix('/').unwrap_or(registry);
let name = &package_key.name;
let version = package_key.suffix.version();
let bare_name = name.bare.as_str();
format!("{registry}/{name}/-/{bare_name}-{version}.tgz")
}
/// A 200 NDJSON response whose body drains the frame channel as the
/// detached resolve task produces frames. Closing the channel (the task
/// dropped its sender) ends the body.