diff --git a/.sqlx/query-052c27b9a93bebeb5db63579da23d02add3df216101e64b26249d5b83c6371ff.json b/.sqlx/query-052c27b9a93bebeb5db63579da23d02add3df216101e64b26249d5b83c6371ff.json new file mode 100644 index 00000000..ca50ff21 --- /dev/null +++ b/.sqlx/query-052c27b9a93bebeb5db63579da23d02add3df216101e64b26249d5b83c6371ff.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM proxy_package_listing WHERE repository_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [] + }, + "hash": "052c27b9a93bebeb5db63579da23d02add3df216101e64b26249d5b83c6371ff" +} diff --git a/.sqlx/query-ac632c79bdae1300865e12ec112fe78d78ab824cf6df7816bdbea4bec2369962.json b/.sqlx/query-ac632c79bdae1300865e12ec112fe78d78ab824cf6df7816bdbea4bec2369962.json new file mode 100644 index 00000000..c268796d --- /dev/null +++ b/.sqlx/query-ac632c79bdae1300865e12ec112fe78d78ab824cf6df7816bdbea4bec2369962.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO proxy_package_listing (repository_id, name, listed_at)\n VALUES ($1, $2, now())\n ON CONFLICT (repository_id, name) DO UPDATE SET listed_at = EXCLUDED.listed_at\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [] + }, + "hash": "ac632c79bdae1300865e12ec112fe78d78ab824cf6df7816bdbea4bec2369962" +} diff --git a/.sqlx/query-b4564dcaebd3ac010815e2747cebaff02205ad3ba5b2e73496df02d4ec18e8ec.json b/.sqlx/query-b4564dcaebd3ac010815e2747cebaff02205ad3ba5b2e73496df02d4ec18e8ec.json new file mode 100644 index 00000000..e4079caf --- /dev/null +++ b/.sqlx/query-b4564dcaebd3ac010815e2747cebaff02205ad3ba5b2e73496df02d4ec18e8ec.json @@ -0,0 +1,43 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT r.id AS \"repository_id!\", l.listed_at AS \"listed_at?\",\n EXISTS (\n SELECT 1 FROM proxy_cached_version v\n WHERE v.repository_id = r.id AND v.name = $2 AND v.version <> '*'\n AND v.confirmed AND NOT v.upstream_missing\n ) AS \"holds!\"\n FROM unnest($1::bigint[]) WITH ORDINALITY AS r(id, ord)\n LEFT JOIN proxy_package_listing l ON l.repository_id = r.id AND l.name = $2\n ORDER BY r.ord\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "repository_id!", + "type_info": "Int8", + "origin": "Expression" + }, + { + "ordinal": 1, + "name": "listed_at?", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "proxy_package_listing", + "name": "listed_at" + } + } + }, + { + "ordinal": 2, + "name": "holds!", + "type_info": "Bool", + "origin": "Expression" + } + ], + "parameters": { + "Left": [ + "Int8Array", + "Text" + ] + }, + "nullable": [ + null, + true, + null + ] + }, + "hash": "b4564dcaebd3ac010815e2747cebaff02205ad3ba5b2e73496df02d4ec18e8ec" +} diff --git a/crates/core/src/repo/proxy.rs b/crates/core/src/repo/proxy.rs index e2476958..8be174e5 100644 --- a/crates/core/src/repo/proxy.rs +++ b/crates/core/src/repo/proxy.rs @@ -127,6 +127,12 @@ //! * **Offline** — `repository.online == false` serves only already-cached data; //! no upstream requests are made, not even to revalidate a stored //! document (a conversation that reaches one serves it whatever its age). +//! * **Cached only** — the web UI's package page reads a proxy as it is +//! offline ([`ProxyRepo::cached_only`]), and its upstream metadata and +//! manifests from the stored documents alone: no upstream request on the +//! page, and nothing discovered by looking. Each upstream listing records +//! when it was answered (`proxy_package_listing`), which the page shows +//! and refreshes by, off the request. use std::collections::{BTreeSet, HashMap, HashSet}; use std::io::Cursor; @@ -563,6 +569,8 @@ pub struct ProxyRepo { /// How long a blob a replaced or deleted stored document named is kept /// before it is deleted (`outbox.grace_seconds`). delete_grace: Duration, + /// Read only what Forge holds ([`Self::cached_only`]). + cached_only: bool, } impl ProxyRepo { @@ -597,9 +605,23 @@ impl ProxyRepo { blobs: None, envelope, delete_grace: Duration::seconds(DEFAULT_DELETE_GRACE_SECONDS), + cached_only: false, } } + /// Read only what Forge holds, as the web UI's package page does (#836): + /// the cached version rows, the stored upstream documents whatever their + /// age, the stored tags and the downloaded packages, as an offline proxy + /// serves them. Nothing is asked of the upstream and nothing is + /// discovered: a package the proxy holds nothing of has no versions. + /// Unlike an offline proxy, the upstream's metadata and manifests are + /// read from the stored documents, so the page has what a client's last + /// request fetched. + pub fn cached_only(mut self) -> Self { + self.cached_only = true; + self + } + /// Use `http` (built from the server's `[proxy]` config) for upstream /// traffic. pub fn with_http(mut self, http: UpstreamHttp) -> Self { @@ -905,8 +927,9 @@ impl ProxyRepo { /// as its freshness allows: through the local cache (never Redis: what /// an upstream answered decides what is downloaded and served) and, when /// the conversation `keep`s its documents, the copy stored for every - /// instance ([`Self::fetch_stored`]). Returns it with the origins its - /// URLs name ([`named_origins`]). + /// instance ([`Self::fetch_stored`]). Read + /// [cached only](Self::cached_only), the stored copy alone, as it is. + /// Returns it with the origins its URLs name ([`named_origins`]). /// /// A 404 is cached only for an immutable document: a metadata 404 (a /// package the upstream does not know yet) is what the not-found @@ -930,38 +953,52 @@ impl ProxyRepo { Some(accept) => format!("{}/{}\n{accept}", Self::upstream_scope(repo), fetch.url), None => format!("{}/{}", Self::upstream_scope(repo), fetch.url), }; - let cached: Result> = self - .cache - .get_or_try_insert_local_with_tags_expiring(NS_UPSTREAM_FETCH, &key, &[], ttl, || { - // On the heap: up to four of these are in flight per round - // of a conversation, each deep in the request's future. - Box::pin(async move { - if keep { - // In a task of its own, from a fresh stack: the - // stored copy's reads and writes (the database, the - // object store) would add their depth to the - // request's, deep already at this point. - let (this, repo, fetch) = (self.clone(), repo.clone(), fetch.clone()); - let url = fetch.url.clone(); - return tokio::spawn( - async move { this.fetch_stored(&repo, &fetch, ttl).await } - .instrument(tracing::Span::current()), - ) - .await - .unwrap_or_else(|e| { - Err(Some(FetchFailure::local(format!("fetch {url}: {e}")))) - }); - } - let cached = match self.request(repo, fetch, None).await? { - Answer::Sent { bytes, named, .. } => CachedFetch::found(&bytes, named), - Answer::NotFound => Self::not_found(fetch)?, - Answer::TooLarge(reason) => CachedFetch::TooLarge { reason }, - Answer::NotModified(_) => return Err(Some(Self::unasked_304(fetch))), - }; - Ok((cached, ttl)) - }) - }) - .await; + let cached: Result> = if self.cached_only { + // Read cached only, the stored copy is read as it is, past the + // local cache: a stale copy must not be kept there for the + // reads that ask the upstream, which share its key, and a fetch + // from the upstream under way there is not waited for. A + // document that is not stored is not fetched. + if keep { + Box::pin(self.fetch_stored_apart(repo, fetch, ttl)) + .await + .map(|(found, _)| found) + } else { + Err(Some(FetchFailure::local(format!( + "{}: not fetched, only what Forge holds is read", + fetch.url + )))) + } + } else { + self.cache + .get_or_try_insert_local_with_tags_expiring( + NS_UPSTREAM_FETCH, + &key, + &[], + ttl, + || { + // On the heap: up to four of these are in flight per round + // of a conversation, each deep in the request's future. + Box::pin(async move { + if keep { + return self.fetch_stored_apart(repo, fetch, ttl).await; + } + let cached = match self.request(repo, fetch, None).await? { + Answer::Sent { bytes, named, .. } => { + CachedFetch::found(&bytes, named) + } + Answer::NotFound => Self::not_found(fetch)?, + Answer::TooLarge(reason) => CachedFetch::TooLarge { reason }, + Answer::NotModified(_) => { + return Err(Some(Self::unasked_304(fetch))); + } + }; + Ok((cached, ttl)) + }) + }, + ) + .await + }; match cached { Ok(CachedFetch::Found { body, named }) => { match base64::engine::general_purpose::STANDARD.decode(body) { @@ -1005,6 +1042,25 @@ impl ProxyRepo { )) } + /// [`Self::fetch_stored`] in a task of its own, from a fresh stack: the + /// stored copy's reads and writes (the database, the object store) would + /// add their depth to the request's, deep already at this point. + async fn fetch_stored_apart( + &self, + repo: &Repository, + fetch: &api::Fetch, + ttl: std::time::Duration, + ) -> Result<(CachedFetch, std::time::Duration), Option> { + let (this, repo, fetch) = (self.clone(), repo.clone(), fetch.clone()); + let url = fetch.url.clone(); + tokio::spawn( + async move { this.fetch_stored(&repo, &fetch, ttl).await } + .instrument(tracing::Span::current()), + ) + .await + .unwrap_or_else(|e| Err(Some(FetchFailure::local(format!("fetch {url}: {e}"))))) + } + /// [`Self::fetch`] past the local cache, through the document stored for /// every instance (`proxy_document`, its bytes in the CAS): within the /// metadata TTL of when the upstream last sent or confirmed it (`ttl`), @@ -1308,8 +1364,11 @@ impl ProxyRepo { } /// Resolve the repository row by name, erroring if missing or not a proxy. + /// Read [cached only](Self::cached_only), the row is taken for an offline + /// one's: everything that would reach the upstream (its index, a stored + /// document's revalidation, a download) serves what is stored instead. async fn proxy_repo(&self, repo: &str) -> Result { - let row = repository_dao::find_repository_by_name(&self.pool, repo) + let mut row = repository_dao::find_repository_by_name(&self.pool, repo) .await? .ok_or_else(|| RepoError::RepoNotFound(repo.to_string()))?; if row.kind() != forge_db::model::RepoKind::Proxy { @@ -1318,6 +1377,9 @@ impl ProxyRepo { kind: row.kind.clone(), }); } + if self.cached_only { + row.online = false; + } Ok(row) } @@ -1351,7 +1413,10 @@ impl ProxyRepo { format!("{}/{lower_id}", Self::upstream_scope(repo)) } - fn metadata_ttl(repo: &Repository) -> Duration { + /// How long `repo` serves what its upstream last listed or sent before + /// asking again (`metadata_ttl_seconds`, default + /// [`DEFAULT_METADATA_TTL_SECONDS`]). + pub fn metadata_ttl(repo: &Repository) -> Duration { Duration::seconds( repo.metadata_ttl_seconds .map(i64::from) @@ -1489,6 +1554,12 @@ impl ProxyRepo { .clone() .ok_or_else(|| RepoError::Storage("proxy repository has no remote_url".into()))?; let key = Self::upstream_scope(repo); + // Read cached only, the persisted index as it is, past the local + // cache: a stale one must not be kept there for the reads that ask + // the upstream. + if self.cached_only { + return self.resolve_index_uncached(repo, ttl, &remote_url).await; + } // The cache factory error must be `Clone` (moka coalescing shares an // `Arc` across waiters); `RepoError` wraps non-`Clone` `sqlx::Error`, @@ -1630,6 +1701,11 @@ impl ProxyRepo { repo: &Repository, lower_id: &str, ) -> Result, RepoError> { + // What is held, as it is: a look at the cache is no lookup of the + // upstream to count. + if self.cached_only { + return self.cached_rows(repo.id, lower_id).await; + } let mut lookup = Lookup::start(&repo.name, LookupKind::Metadata); let span = lookup.span(); let result = self @@ -1807,6 +1883,11 @@ impl ProxyRepo { ) .await?; } + // When, for the web UI's package page (best effort: it only + // decides how old the page says it is, and when it refreshes). + if let Err(e) = proxy_dao::record_listing(&self.pool, repo.id, lower_id).await { + tracing::warn!(repo = %repo.name, id = %lower_id, error = %e, "recording when the upstream listed the package failed"); + } let applied = self.package_delay(repo).await?.for_package(lower_id); if applied.delay_seconds > 0 { self.date_delay_from_upstream(repo, &applied, lower_id) @@ -2627,7 +2708,9 @@ impl ProxyRepo { /// failure yield nothing, and the caller falls back to the listing's /// view (listed, no dependencies). Records the package id's upstream /// casing for search (best effort): the newest version's, the one the - /// upstream shows, as the format reports it. + /// upstream shows, as the format reports it. Read + /// [cached only](Self::cached_only), the metadata comes from the stored + /// documents alone, and a document not stored is no failure to report. async fn upstream_metadata( &self, repo: &Repository, @@ -2635,9 +2718,16 @@ impl ProxyRepo { wanted: &[String], dating: bool, ) -> HashMap { - if !repo.online || wanted.is_empty() { + if (!repo.online && !self.cached_only) || wanted.is_empty() { return HashMap::new(); } + let unavailable = |reason: &dyn std::fmt::Display| { + if self.cached_only { + tracing::debug!(repo = %repo.name, id = %lower_id, %reason, "no stored upstream metadata; serving the listing's"); + } else { + tracing::warn!(repo = %repo.name, id = %lower_id, %reason, "upstream metadata unavailable; serving the listing's"); + } + }; let versions = api::PackageVersions { id: lower_id.to_string(), versions: wanted.to_vec(), @@ -2650,7 +2740,7 @@ impl ProxyRepo { let metadata = match Box::pin(self.converse_resolved(repo, op)).await { Ok(api::UpstreamAnswer::Metadata(metadata)) => metadata, Ok(api::UpstreamAnswer::Failed(reason)) => { - tracing::warn!(repo = %repo.name, id = %lower_id, %reason, "upstream metadata unavailable; serving the listing's"); + unavailable(&reason); return HashMap::new(); } Ok(_) => { @@ -2658,7 +2748,7 @@ impl ProxyRepo { return HashMap::new(); } Err(e) => { - tracing::warn!(repo = %repo.name, id = %lower_id, error = %e, "upstream metadata unavailable; serving the listing's"); + unavailable(&e); return HashMap::new(); } }; @@ -3296,6 +3386,16 @@ impl RepoReadOps for ProxyRepo { if !known { return Ok(None); } + if self.cached_only { + // The upstream's manifest as it is stored, when a client's + // request fetched it; nothing is downloaded. + return Ok(Box::pin(self.upstream_manifest(&repo_row, &lower, &norm)) + .await + .unwrap_or_else(|e| { + tracing::debug!(repo = %repo_row.name, id = %lower, version = %norm, error = %e, "no stored upstream manifest"); + None + })); + } if repo_row.online { if let Some(bytes) = Box::pin(self.upstream_manifest(&repo_row, &lower, &norm)).await? { return Ok(Some(bytes)); diff --git a/crates/db/src/proxy_dao.rs b/crates/db/src/proxy_dao.rs index 700232ce..9a665900 100644 --- a/crates/db/src/proxy_dao.rs +++ b/crates/db/src/proxy_dao.rs @@ -1,4 +1,5 @@ -//! Data access for `proxy_cached_version` and `proxy_upstream_index`. +//! Data access for `proxy_cached_version`, `proxy_package_listing` and +//! `proxy_upstream_index`. use serde_json::Value as JsonValue; use sqlx::{PgConnection, PgPool, Postgres, Transaction}; @@ -387,8 +388,10 @@ where /// missing-upstream flags cleared and their metadata due for a refresh from /// the new upstream. The versions released from the package delay early are /// forgotten too: they were decisions about the previous upstream's packages, -/// and the new upstream's package of the same name may be another one. -/// Returns the number of deleted cached-version rows. +/// and the new upstream's package of the same name may be another one. So +/// are the times the previous upstream listed each package +/// (`proxy_package_listing`). Returns the number of deleted cached-version +/// rows. pub async fn reset_for_new_upstream( tx: &mut Transaction<'_, Postgres>, repository_id: i64, @@ -423,9 +426,83 @@ pub async fn reset_for_new_upstream( ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM proxy_package_listing WHERE repository_id = $1", + repository_id, + ) + .execute(&mut **tx) + .await?; Ok(deleted) } +/// Record that the upstream answered the version listing of package `name` +/// now (`proxy_package_listing`). +pub async fn record_listing(pool: &PgPool, repository_id: i64, name: &str) -> sqlx::Result<()> { + sqlx::query!( + r#" + INSERT INTO proxy_package_listing (repository_id, name, listed_at) + VALUES ($1, $2, now()) + ON CONFLICT (repository_id, name) DO UPDATE SET listed_at = EXCLUDED.listed_at + "#, + repository_id, + name, + ) + .execute(pool) + .await?; + Ok(()) +} + +/// What one proxy holds of a package ([`listing_states`]). +#[derive(Debug, Clone)] +pub struct ListingState { + pub repository_id: i64, + /// When the upstream last answered the package's version listing, if it + /// ever did since it was recorded. + pub listed_at: Option>, + /// Whether the proxy holds an upstream-confirmed version of the package + /// that the upstream still lists (what its read path enumerates, before + /// the package delay and the not-found markers). + pub holds: bool, +} + +/// For each of the proxies `repository_ids`, when its upstream last listed +/// package `name` and whether it holds any version of it, in the order +/// given. +pub async fn listing_states( + pool: &PgPool, + repository_ids: &[i64], + name: &str, +) -> sqlx::Result> { + if repository_ids.is_empty() { + return Ok(Vec::new()); + } + let rows = sqlx::query!( + r#" + SELECT r.id AS "repository_id!", l.listed_at AS "listed_at?", + EXISTS ( + SELECT 1 FROM proxy_cached_version v + WHERE v.repository_id = r.id AND v.name = $2 AND v.version <> '*' + AND v.confirmed AND NOT v.upstream_missing + ) AS "holds!" + FROM unnest($1::bigint[]) WITH ORDINALITY AS r(id, ord) + LEFT JOIN proxy_package_listing l ON l.repository_id = r.id AND l.name = $2 + ORDER BY r.ord + "#, + repository_ids, + name, + ) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map(|row| ListingState { + repository_id: row.repository_id, + listed_at: row.listed_at, + holds: row.holds, + }) + .collect()) +} + /// Forget the upstream 404s a proxy remembers of package `name`, or of every /// package (`None`), inside the transaction of an administrator's cache /// invalidation, so the next request asks the upstream again. Package-level diff --git a/crates/server/src/browse.rs b/crates/server/src/browse.rs index 67f96961..429e0494 100644 --- a/crates/server/src/browse.rs +++ b/crates/server/src/browse.rs @@ -12,9 +12,10 @@ //! evaluated against the request's [`Principal`]; //! * version visibility, group member resolution, listed/prerelease filtering, //! and proxy **package-delay** gating all come for free from -//! [`AppState::read_ops_for`] + [`RepoReadOps`] — the same ops the repository -//! dispatcher uses, so browse can never surface a version the protocol would -//! hide. +//! [`AppState::read_ops_for`] (the package detail's +//! [`AppState::cached_read_ops_for`]) + [`RepoReadOps`] — the same ops the +//! repository dispatcher uses, so browse can never surface a version the +//! protocol would hide. //! //! ## Endpoints //! * `GET /browse/repos` — the repositories the caller may read (the SPA's @@ -26,6 +27,12 @@ //! mode remains for API callers. //! * `GET /browse/packages/{repo}/{id}` — package detail: visible versions with //! per-version metadata + dependency groups parsed from the stored `.nuspec`. +//! It reads what Forge holds: a proxy (a group's proxy members) asks its +//! upstream nothing on the request, says when the upstream last listed the +//! package, and refreshes it off the request once that is stale (#836). +//! * `POST /browse/packages/{repo}/{id}/refresh` — **Refresh from upstream**: +//! the package asked of the upstream now, as a client's request does, then +//! its detail. //! //! An offline hosted or group repository serves nothing here either: its //! package search and detail answer 503 with `Retry-After`, as its feed does, @@ -39,6 +46,9 @@ //! no `/stats` endpoint. The DTOs keep room for them to be added without //! breaking the SPA. `published_at` maps to `component.created_at`. +use std::collections::HashMap; +use std::sync::Arc; + use axum::extract::{Path, Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Json, Response}; @@ -46,10 +56,12 @@ use axum::routing::get; use axum::Router; use chrono::{DateTime, Utc}; use forge_core::auth::principal::Principal; +use forge_core::format::Format; use forge_core::repo::delay::{AppliedDelay, PackageDelay, PackageGate}; -use forge_core::repo::SearchParams; +use forge_core::repo::proxy::ProxyRepo; +use forge_core::repo::{RegistrationEntry, SearchParams}; use forge_db::model::{RepoKind, Repository}; -use forge_db::{component_dao, repository_dao}; +use forge_db::{component_dao, proxy_dao, repository_dao}; use forge_plugin::api::Storage; use serde::{Deserialize, Serialize}; @@ -441,63 +453,160 @@ pub struct PackageDetail { /// other kinds, and for proxies of formats without a package delay. #[serde(skip_serializing_if = "Option::is_none")] pub package_delay: Option, + /// When the upstream last listed the package's versions, for a proxy; + /// for a group, the oldest such time among the proxies serving it. What + /// the page shows is what Forge held then (and what clients fetched + /// since). Absent for a hosted repository, and when no time was + /// recorded. + #[serde(skip_serializing_if = "Option::is_none")] + pub cached_at: Option>, + /// Whether a refresh from the upstream is under way off the request: a + /// look again shortly shows what it found. + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub refreshing: bool, } -/// `GET /browse/packages/{repo}/{id}` — package detail. -pub async fn package_detail( - State(state): State, - Path((repo_name, id)): Path<(String, String)>, - req: axum::extract::Request, -) -> Response { - let principal = principal_of(&req); - if let Some(denied) = require_user(&principal) { - return denied; +/// The package a detail or refresh request names, once the caller may read +/// it: its repository, format and lowered id; else the response to send. +async fn package_target( + state: &AppState, + principal: &Principal, + repo_name: &str, + id: &str, +) -> Result<(Repository, Arc, String), Response> { + if let Some(denied) = require_user(principal) { + return Err(denied); } - let repo = match repository_dao::find_repository_by_name(&state.pool, &repo_name).await { + let repo = match repository_dao::find_repository_by_name(&state.pool, repo_name).await { // An images (Docker) repository holds images (`/browse/images`) and // a files (raw) one files (its listings), not packages. Ok(Some(r)) if state.storage(&r.format) == Storage::Packages => r, // An unknown name answers like an unreadable repository (no name // enumeration), except to a global admin. - Ok(_) if state.authz.is_global_admin(&principal).await => { - return (StatusCode::NOT_FOUND, "repository not found").into_response(); + Ok(_) if state.authz.is_global_admin(principal).await => { + return Err((StatusCode::NOT_FOUND, "repository not found").into_response()); } - Ok(_) => return (StatusCode::FORBIDDEN, "forbidden").into_response(), - Err(e) => return db_error(e), + Ok(_) => return Err((StatusCode::FORBIDDEN, "forbidden").into_response()), + Err(e) => return Err(db_error(e)), }; // Per-repo read authz; the caller is a signed-in user at this point, so a // denial is 403 (401 for anonymous callers was already returned above). - if !state.authz.can_read(&principal, &repo).await { - return (StatusCode::FORBIDDEN, "forbidden").into_response(); + if !state.authz.can_read(principal, &repo).await { + return Err((StatusCode::FORBIDDEN, "forbidden").into_response()); } // An offline hosted or group repository serves nothing, as on its feed. if offline_serves_nothing(&repo) { - return offline_response(); + return Err(offline_response()); } - let format = match state.formats.for_repository(&repo) { - Ok(format) => format, - Err(e) => return read_err(e), - }; - let lower = format.lower_id(&id); + let format = state.formats.for_repository(&repo).map_err(read_err)?; + let lower = format.lower_id(id); if !state .authz - .package_filter(&principal, &repo) + .package_filter(principal, &repo) .await .allows(&lower) { - return (StatusCode::NOT_FOUND, "package not found").into_response(); + return Err((StatusCode::NOT_FOUND, "package not found").into_response()); } + Ok((repo, format, lower)) +} + +/// `GET /browse/packages/{repo}/{id}` — package detail, from what Forge holds +/// (#836). +/// +/// A proxy, and a group's proxy members, are read +/// [cached only](forge_core::repo::proxy::ProxyRepo::cached_only): the request +/// asks no upstream, whatever the age of what is cached, and a package the +/// proxy holds nothing of is not found (and not discovered by looking). The +/// detail says when the upstream last listed the package (`cached_at`). Once +/// that is past the proxy's metadata TTL, or older than its last edit or +/// cache invalidation, the page starts a refresh off the request +/// ([`BackgroundRefreshes`]), and says so (`refreshing`): the next look shows +/// what the upstream answered. +pub async fn package_detail( + State(state): State, + Path((repo_name, id)): Path<(String, String)>, + req: axum::extract::Request, +) -> Response { + let principal = principal_of(&req); + let (repo, format, lower) = match package_target(&state, &principal, &repo_name, &id).await { + Ok(target) => target, + Err(response) => return response, + }; + let listing = listing_of(&state, &repo, &format, &principal, &lower).await; + let refreshing = state + .browse_refreshes + .start(&state, listing.due, &principal, &lower); + cached_detail( + &state, + &repo, + &format, + principal, + &id, + &lower, + listing.cached_at, + refreshing, + ) + .await +} - let ops = match state.read_ops_for(&repo, principal).await { +/// `POST /browse/packages/{repo}/{id}/refresh` — **Refresh from upstream**: +/// ask the upstream for the package now, as a client's request does +/// ([`refresh_from_upstream`]), then answer its detail as +/// [`package_detail`] does. Every reader of the repository may: it costs +/// the upstream what a client's install does, and what is still within the +/// metadata TTL is not asked again. A package the proxy held nothing of is +/// discovered this way, when the upstream has it. +pub async fn refresh_package( + State(state): State, + Path((repo_name, id)): Path<(String, String)>, + req: axum::extract::Request, +) -> Response { + let principal = principal_of(&req); + let (repo, format, lower) = match package_target(&state, &principal, &repo_name, &id).await { + Ok(target) => target, + Err(response) => return response, + }; + if let Err(e) = refresh_from_upstream(&state, &repo, &format, principal.clone(), &lower).await { + return read_err(e); + } + let listing = listing_of(&state, &repo, &format, &principal, &lower).await; + cached_detail( + &state, + &repo, + &format, + principal, + &id, + &lower, + listing.cached_at, + false, + ) + .await +} + +/// The detail of package `lower` (`id` as requested) of `repo`, read from +/// what Forge holds ([`AppState::cached_read_ops_for`]). +#[allow(clippy::too_many_arguments)] +async fn cached_detail( + state: &AppState, + repo: &Repository, + format: &Arc, + principal: Principal, + id: &str, + lower: &str, + cached_at: Option>, + refreshing: bool, +) -> Response { + let ops = match state.cached_read_ops_for(repo, principal).await { Ok(o) => o, Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response(), }; // Visible versions (delay-gated + group-resolved by the read ops). - let versions = match ops.list_versions(&repo.name, &lower).await { + let versions = match ops.list_versions(&repo.name, lower).await { Ok(v) => v, Err(e) => return read_err(e), }; @@ -510,7 +619,7 @@ pub async fn package_detail( // package, so we must not 404 it just because `versions` is empty. let is_proxy = repo.kind() == RepoKind::Proxy; let gate = if is_proxy { - match package_gate(&state, &repo, &lower).await { + match package_gate(state, repo, lower).await { Ok(gate) => Some(gate), Err(e) => return db_error(e), } @@ -518,7 +627,7 @@ pub async fn package_detail( None }; let delayed = match &gate { - Some(gate) => match delayed_proxy_versions(&state, &repo, &lower, gate).await { + Some(gate) => match delayed_proxy_versions(state, repo, lower, gate).await { Ok(d) => d, Err(e) => return db_error(e), }, @@ -531,7 +640,7 @@ pub async fn package_detail( // Registration entries carry the stored manifest summary (or, for older // versions, the `.nuspec` bytes) for deps. - let registration = match ops.registration(&repo.name, &lower).await { + let registration = match ops.registration(&repo.name, lower).await { Ok(r) => r, Err(e) => return read_err(e), }; @@ -542,7 +651,7 @@ pub async fn package_detail( let original_id = versions .first() .map(|v| v.original_id.clone()) - .unwrap_or_else(|| id.clone()); + .unwrap_or_else(|| id.to_string()); let mut dependency_groups_by_version = std::collections::BTreeMap::new(); let mut deprecated = std::collections::BTreeMap::new(); @@ -582,7 +691,7 @@ pub async fn package_detail( // Per-version size / published_at from the local component rows when // present (one query for every version, not two per version). let local: std::collections::HashMap, Option)> = - component_dao::version_sizes(&state.pool, repo.id, &lower, format.package_asset_path()) + component_dao::version_sizes(&state.pool, repo.id, lower, format.package_asset_path()) .await .unwrap_or_else(|e| { tracing::warn!(error = %e, "browse: version sizes lookup failed"); @@ -627,19 +736,11 @@ pub async fn package_detail( // the UI can render a delay badge alongside any downloadable versions. detail_versions.extend(delayed); - let tags = ops.tags(&repo.name, &lower).await.unwrap_or_else(|e| { + let tags = ops.tags(&repo.name, lower).await.unwrap_or_else(|e| { tracing::warn!(error = %e, "browse: tags lookup failed"); Default::default() }); - // The latest version: the one tagged so, else the highest. - let latest = tags - .get("latest") - .and_then(|l| registration.iter().find(|e| &e.version == l)) - .or_else(|| { - registration - .iter() - .max_by(|a, b| format.compare_versions(&a.version, &b.version)) - }); + let latest = latest_entry(format.as_ref(), &tags, ®istration); let description = latest.and_then(|entry| { let summary = entry.manifest_summary.as_ref(); summary @@ -654,7 +755,7 @@ pub async fn package_detail( .filter(|d| !d.trim().is_empty()) }); let readme = match latest { - Some(entry) => match ops.resolve_nuspec(&repo.name, &lower, &entry.version).await { + Some(entry) => match ops.resolve_nuspec(&repo.name, lower, &entry.version).await { Ok(Some(manifest)) => { let format = format.clone(); tokio::task::spawn_blocking(move || format.readme(&manifest)) @@ -674,7 +775,7 @@ pub async fn package_detail( Json(PackageDetail { repo: repo.name.clone(), repo_kind: repo.kind.clone(), - id: lower, + id: lower.to_string(), original_id, versions: detail_versions, dependency_groups_by_version, @@ -684,13 +785,255 @@ pub async fn package_detail( readme, package_delay: gate .filter(|_| { - crate::admin::repositories::without_package_delay(&state, &repo.format).is_none() + crate::admin::repositories::without_package_delay(state, &repo.format).is_none() }) .map(|g| g.applied), + cached_at, + refreshing, }) .into_response() } +/// The latest of a package's `registration` entries: the one its `latest` +/// tag names, else the highest version. +fn latest_entry<'a>( + format: &dyn Format, + tags: &std::collections::BTreeMap, + registration: &'a [RegistrationEntry], +) -> Option<&'a RegistrationEntry> { + tags.get("latest") + .and_then(|l| registration.iter().find(|e| &e.version == l)) + .or_else(|| { + registration + .iter() + .max_by(|a, b| format.compare_versions(&a.version, &b.version)) + }) +} + +/// Ask `repo`'s upstream (a group's: its serving members') for package +/// `lower`, through the read path a client's request takes, and keep what it +/// answers for the package page: the version listing, the versions' +/// metadata, the tags and the latest version's manifest (npm's README). What +/// is still within the metadata TTL is served from the cache, not asked +/// again, as for a client. Only the listing's failure is one: the rest is +/// best effort, as on the page. +async fn refresh_from_upstream( + state: &AppState, + repo: &Repository, + format: &Arc, + principal: Principal, + lower: &str, +) -> Result<(), forge_core::repo::RepoError> { + let ops = state.read_ops_for(repo, principal).await?; + ops.list_versions(&repo.name, lower).await?; + let registration = ops + .registration(&repo.name, lower) + .await + .unwrap_or_else(|e| { + tracing::warn!(repo = %repo.name, id = %lower, error = %e, "browse: refreshing the registration failed"); + Vec::new() + }); + let tags = ops.tags(&repo.name, lower).await.unwrap_or_else(|e| { + tracing::warn!(repo = %repo.name, id = %lower, error = %e, "browse: refreshing the tags failed"); + Default::default() + }); + if let Some(latest) = latest_entry(format.as_ref(), &tags, ®istration) { + if let Err(e) = ops.resolve_nuspec(&repo.name, lower, &latest.version).await { + tracing::warn!(repo = %repo.name, id = %lower, error = %e, "browse: refreshing the latest manifest failed"); + } + } + Ok(()) +} + +/// How old what a proxy, or a group's serving proxies, hold of a package is +/// ([`listing_of`]). +#[derive(Default)] +struct Listing { + /// When the upstream last listed the package's versions; for a group, + /// the oldest such time among its serving proxies that hold the package. + /// `None` when none of them recorded one. + cached_at: Option>, + /// The online proxies holding the package whose listing is due for a + /// refresh: never recorded, past the proxy's metadata TTL, or older than + /// its last edit or cache invalidation (`updated_at`). + due: Vec, +} + +/// The [`Listing`] of package `lower` of `repo`: a proxy's own, a group's +/// over the proxies serving the package to `principal`, nothing for a hosted +/// repository. A proxy that holds nothing of the package counts for neither: +/// looking at a page discovers nothing. Best effort: a failure to tell +/// leaves the page without an age, and refreshes nothing. +async fn listing_of( + state: &AppState, + repo: &Repository, + format: &Arc, + principal: &Principal, + lower: &str, +) -> Listing { + let proxies: Vec = match repo.kind() { + RepoKind::Hosted => return Listing::default(), + RepoKind::Proxy => vec![repo.clone()], + RepoKind::Group => match state + .group_repo(principal.clone(), true) + .with_format(format.clone()) + .serving_leaves(&repo.name, lower) + .await + { + Ok(leaves) => leaves + .into_iter() + .filter(|leaf| leaf.kind() == RepoKind::Proxy) + .collect(), + Err(e) => { + tracing::warn!(repo = %repo.name, id = %lower, error = %e, "browse: the group's serving members could not be resolved"); + return Listing::default(); + } + }, + }; + if proxies.is_empty() { + return Listing::default(); + } + let ids: Vec = proxies.iter().map(|proxy| proxy.id).collect(); + let (states, now) = match tokio::try_join!( + proxy_dao::listing_states(&state.pool, &ids, lower), + forge_db::db_now(&state.pool), + ) { + Ok(found) => found, + Err(e) => { + tracing::warn!(repo = %repo.name, id = %lower, error = %e, "browse: when the upstream listed the package could not be read"); + return Listing::default(); + } + }; + let mut listing = Listing::default(); + for (proxy, held) in proxies.into_iter().zip(states) { + if !held.holds { + continue; + } + if let Some(at) = held.listed_at { + listing.cached_at = Some(listing.cached_at.map_or(at, |oldest| oldest.min(at))); + } + let due = held + .listed_at + .is_none_or(|at| at + ProxyRepo::metadata_ttl(&proxy) <= now || at < proxy.updated_at); + if due && proxy.online { + listing.due.push(proxy); + } + } + listing +} + +/// At most this many package-page refreshes run off their requests at once +/// on an instance; a page that finds none free refreshes nothing. +const MAX_BACKGROUND_REFRESHES: usize = 4; +/// A proxy's package is refreshed off a page's request at most once within +/// this long, whatever the refresh found: a page left open, or an upstream +/// that keeps failing, costs one refresh a minute at most. +const BACKGROUND_REFRESH_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(60); +/// The refreshes remembered before those past their cooldown are let go. +const BACKGROUND_REFRESH_MEMORY: usize = 10_000; + +/// The package pages' refreshes from the upstream off their requests +/// (#836), on one instance: one at a time per proxy and package, at most +/// [`MAX_BACKGROUND_REFRESHES`] at once, and none again within +/// [`BACKGROUND_REFRESH_COOLDOWN`] of the last one. +pub struct BackgroundRefreshes { + /// Per (proxy id, package): `None` while its refresh runs, else when the + /// last one ended. + seen: std::sync::Mutex>>, + permits: Arc, +} + +impl Default for BackgroundRefreshes { + fn default() -> Self { + BackgroundRefreshes { + seen: Default::default(), + permits: Arc::new(tokio::sync::Semaphore::new(MAX_BACKGROUND_REFRESHES)), + } + } +} + +impl BackgroundRefreshes { + /// Refresh package `lower` of each of the `due` proxies off the request + /// ([`refresh_from_upstream`], for `principal`), unless its refresh is + /// under way, one ended within the cooldown, or the instance runs as + /// many as it may. Whether one of them is under way now. + pub fn start( + self: &Arc, + state: &AppState, + due: Vec, + principal: &Principal, + lower: &str, + ) -> bool { + let mut refreshing = false; + for proxy in due { + let key = (proxy.id, lower.to_string()); + let permit = { + let mut seen = self.seen(); + match seen.get(&key) { + Some(None) => { + refreshing = true; + continue; + } + Some(Some(ended)) if ended.elapsed() < BACKGROUND_REFRESH_COOLDOWN => continue, + _ => {} + } + let Ok(permit) = self.permits.clone().try_acquire_owned() else { + continue; + }; + if seen.len() >= BACKGROUND_REFRESH_MEMORY { + seen.retain(|_, ended| { + ended.is_none_or(|ended| ended.elapsed() < BACKGROUND_REFRESH_COOLDOWN) + }); + } + seen.insert(key.clone(), None); + permit + }; + refreshing = true; + let ended = RefreshEnded { + refreshes: self.clone(), + key, + }; + let (state, principal, lower) = (state.clone(), principal.clone(), lower.to_string()); + tokio::spawn(async move { + let (_permit, _ended) = (permit, ended); + let refreshed = match state.formats.for_repository(&proxy) { + Ok(format) => { + refresh_from_upstream(&state, &proxy, &format, principal, &lower).await + } + Err(e) => Err(e), + }; + if let Err(e) = refreshed { + tracing::warn!(repo = %proxy.name, id = %lower, error = %e, "browse: refreshing the package from the upstream failed"); + } + }); + } + refreshing + } + + fn seen( + &self, + ) -> std::sync::MutexGuard<'_, HashMap<(i64, String), Option>> { + self.seen + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +/// Marks a background refresh ended when its task does, however it ends. +struct RefreshEnded { + refreshes: Arc, + key: (i64, String), +} + +impl Drop for RefreshEnded { + fn drop(&mut self) { + self.refreshes.seen().insert( + std::mem::take(&mut self.key), + Some(std::time::Instant::now()), + ); + } +} + /// The package delay of proxy package `lower_id` and its versions released /// early. A failure to resolve the delay is a database error. async fn package_gate( @@ -798,6 +1141,10 @@ pub fn browse_router() -> Router { .route("/browse/repos", get(list_repos)) .route("/browse/packages", get(search_packages)) .route("/browse/packages/{repo}/{id}", get(package_detail)) + .route( + "/browse/packages/{repo}/{id}/refresh", + axum::routing::post(refresh_package), + ) .route("/browse/images", get(crate::browse_images::list_images)) .route("/browse/chart", get(crate::browse_images::chart_detail)) .route( diff --git a/crates/server/src/state.rs b/crates/server/src/state.rs index 50efd715..d4d87f6b 100644 --- a/crates/server/src/state.rs +++ b/crates/server/src/state.rs @@ -118,6 +118,9 @@ pub struct AppState { /// The email service (`/admin/settings/email`): sends invitations and /// test messages with the settings stored in the database. pub mailer: Arc, + /// The package pages' refreshes from the upstream under way off their + /// requests, on this instance (#836). + pub browse_refreshes: Arc, } /// One instance of Forge: each replica of an install has state of its own @@ -412,6 +415,7 @@ impl AppState { )), assertion_issuer, envelope, + browse_refreshes: Arc::default(), } } @@ -524,20 +528,46 @@ impl AppState { .with_download_tally(self.downloads.clone()), ), RepoKind::Proxy => Arc::new(self.proxy_repo(format)), - RepoKind::Group => Arc::new( - GroupRepo::new( - self.pool.clone(), - principal, - Arc::new(self.member_factory()), - self.authz.clone(), - self.config.groups.max_nesting_depth.max(0) as u32, - ) - .with_format(format), - ), + RepoKind::Group => Arc::new(self.group_repo(principal, false).with_format(format)), + }; + Ok(Arc::new(PackageFilteredReadOps::new(inner, filter))) + } + + /// [`Self::read_ops_for`] reading only what Forge holds, as the web UI's + /// package page does (#836): a proxy, and a group's proxy members, are + /// read [cached only](ProxyRepo::cached_only), without a request to + /// their upstream. A hosted repository reads as it always does. + pub async fn cached_read_ops_for( + &self, + repo: &Repository, + principal: Principal, + ) -> Result, RepoError> { + let format = self.formats.for_repository(repo)?; + let inner: Arc = match repo.kind() { + RepoKind::Hosted => return self.read_ops_for(repo, principal).await, + RepoKind::Proxy => Arc::new(self.proxy_repo(format).cached_only()), + RepoKind::Group => { + Arc::new(self.group_repo(principal.clone(), true).with_format(format)) + } }; + let filter = self.authz.package_filter(&principal, repo).await; Ok(Arc::new(PackageFilteredReadOps::new(inner, filter))) } + /// A group resolver for `principal`, its proxy members read + /// [cached only](ProxyRepo::cached_only) when `cached_only` is set. + pub fn group_repo(&self, principal: Principal, cached_only: bool) -> GroupRepo { + let mut factory = self.member_factory(); + factory.cached_only = cached_only; + GroupRepo::new( + self.pool.clone(), + principal, + Arc::new(factory), + self.authz.clone(), + self.config.groups.max_nesting_depth.max(0) as u32, + ) + } + /// A fully wired proxy read facade: shared upstream client, local blob /// access, and the envelope service for encrypted upstream passwords. pub fn proxy_repo(&self, format: Arc) -> ProxyRepo { @@ -566,6 +596,7 @@ impl AppState { upstream: self.upstream.clone(), downloads: self.downloads.clone(), delete_grace: self.config.outbox.grace(), + cached_only: false, } } } @@ -648,6 +679,9 @@ pub struct MemberFactory { /// `outbox.grace_seconds`, for the blobs of the proxies' replaced /// upstream documents. delete_grace: chrono::Duration, + /// Proxy members read only what Forge holds + /// ([`ProxyRepo::cached_only`]). + cached_only: bool, } impl MemberOpsFactory for MemberFactory { @@ -664,8 +698,8 @@ impl MemberOpsFactory for MemberFactory { ) .with_download_tally(self.downloads.clone()), )), - RepoKind::Proxy => Ok(Arc::new( - ProxyRepo::new( + RepoKind::Proxy => { + let proxy = ProxyRepo::new( self.pool.clone(), format, self.cas.clone(), @@ -677,8 +711,13 @@ impl MemberOpsFactory for MemberFactory { store: self.blob_store.clone(), })) .with_download_tally(self.downloads.clone()) - .with_delete_grace(self.delete_grace), - )), + .with_delete_grace(self.delete_grace); + Ok(Arc::new(if self.cached_only { + proxy.cached_only() + } else { + proxy + })) + } RepoKind::Group => Err(RepoError::WrongKind { name: member.name.clone(), kind: member.kind.clone(), diff --git a/crates/server/tests/browse.rs b/crates/server/tests/browse.rs index 1a7cd694..8b0d6679 100644 --- a/crates/server/tests/browse.rs +++ b/crates/server/tests/browse.rs @@ -4,11 +4,21 @@ //! the signed-in-users-only rule (anonymous browse is 401 even for //! `anonymous_read` repos — those stay readable on their repository feed only); //! the `/browse/repos` folder root; group-repo merge; dependency-group -//! rendering; and that mounting `/browse` + `/admin/me` did not shadow the -//! admin-only `/admin/*` API. +//! rendering; a proxy's and a group's package page read from the cache, +//! refreshed off the request and on Refresh (#836); and that mounting +//! `/browse` + `/admin/me` did not shadow the admin-only `/admin/*` API. mod common; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::routing::get; +use axum::{Json, Router}; use common::ServerOpts; use serde_json::{json, Value}; @@ -283,6 +293,17 @@ async fn browse_requires_login() { let resp = srv.client().get(&url).send().await.unwrap(); assert_eq!(resp.status(), 401, "anonymous {url} must be 401"); } + // Refresh from upstream too (#836). + let resp = srv + .client() + .post(format!( + "{}/browse/packages/{pub_repo}/{pub_id}/refresh", + srv.base_url + )) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 401, "anonymous refresh must be 401"); // The same anonymous caller can still read the repo through the NuGet feed. let resp = srv @@ -869,3 +890,528 @@ async fn admin_api_not_shadowed_by_new_routes() { .unwrap(); assert_eq!(resp.status(), 200); } + +/// A NuGet upstream for the package page's tests (#836): a flat container, a +/// registration with each version's dependency on `Dep.Lib`, and the `.nuspec` and +/// `.nupkg` of its versions. It records the path of every request, and holds +/// them all while it is closed. +struct Upstream { + base: String, + /// Package id as published → its versions. + packages: Mutex>>, + requests: Mutex>, + open: tokio::sync::watch::Sender, +} + +impl Upstream { + async fn start(packages: &[(&str, &str)]) -> Arc { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind upstream"); + let upstream = Arc::new(Upstream { + base: format!("http://{}", listener.local_addr().unwrap()), + packages: Mutex::new(HashMap::new()), + requests: Mutex::new(Vec::new()), + open: tokio::sync::watch::channel(true).0, + }); + for (id, version) in packages { + upstream.publish(id, version); + } + let app = Router::new() + .route("/index.json", get(upstream_index)) + .route("/flat/{id}/index.json", get(upstream_versions)) + .route("/flat/{id}/{version}/{file}", get(upstream_file)) + .route("/registration/{id}/index.json", get(upstream_registration)) + .with_state(upstream.clone()); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve upstream"); + }); + upstream + } + + fn publish(&self, id: &str, version: &str) { + self.packages + .lock() + .unwrap() + .entry(id.to_string()) + .or_default() + .push(version.to_string()); + } + + /// The package published as `lower` (any casing), with its versions. + fn package(&self, lower: &str) -> Option<(String, Vec)> { + self.packages + .lock() + .unwrap() + .iter() + .find(|(id, _)| id.to_lowercase() == lower) + .map(|(id, versions)| (id.clone(), versions.clone())) + } + + /// Hold every request until [`Self::open`] (`false`), or let them through. + fn set_open(&self, open: bool) { + self.open.send_replace(open); + } + + /// Every request so far. + fn total(&self) -> usize { + self.requests.lock().unwrap().len() + } + + /// Wait (ten seconds at most) until it has had more than `seen` requests. + async fn asked_beyond(&self, seen: usize) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while self.total() <= seen { + assert!(tokio::time::Instant::now() < deadline, "never asked"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + /// The requests that named package `lower`. + fn requests_for(&self, lower: &str) -> usize { + self.requests + .lock() + .unwrap() + .iter() + .filter(|path| path.contains(&format!("/{lower}/"))) + .count() + } + + async fn record(&self, path: String) { + self.requests.lock().unwrap().push(path); + let mut open = self.open.subscribe(); + let _ = open.wait_for(|open| *open).await; + } +} + +async fn upstream_index(State(up): State>) -> Json { + up.record("/index.json".into()).await; + Json(json!({ + "version": "3.0.0", + "resources": [ + {"@id": format!("{}/flat/", up.base), "@type": "PackageBaseAddress/3.0.0"}, + {"@id": format!("{}/registration/", up.base), "@type": "RegistrationsBaseUrl/3.6.0"}, + ], + })) +} + +async fn upstream_versions(State(up): State>, Path(id): Path) -> Response { + up.record(format!("/flat/{id}/index.json")).await; + match up.package(&id) { + Some((_, versions)) => Json(json!({"versions": versions})).into_response(), + None => StatusCode::NOT_FOUND.into_response(), + } +} + +async fn upstream_registration( + State(up): State>, + Path(id): Path, +) -> Response { + up.record(format!("/registration/{id}/index.json")).await; + let Some((original, versions)) = up.package(&id) else { + return StatusCode::NOT_FOUND.into_response(); + }; + let leaves: Vec = versions + .iter() + .map(|v| { + json!({ + "@id": format!("{}/registration/{id}/{v}.json", up.base), + "catalogEntry": { + "id": original, + "version": v, + "listed": true, + "dependencyGroups": [{ + "targetFramework": "net8.0", + "dependencies": [{"id": "Dep.Lib", "range": "[1.0.0, )"}], + }], + }, + "packageContent": format!("{}/flat/{id}/{v}/{id}.{v}.nupkg", up.base), + }) + }) + .collect(); + Json(json!({ + "count": 1, + "items": [{ + "@id": format!("{}/registration/{id}/index.json#page", up.base), + "lower": versions.first(), + "upper": versions.last(), + "count": leaves.len(), + "items": leaves, + }], + })) + .into_response() +} + +async fn upstream_file( + State(up): State>, + Path((id, version, file)): Path<(String, String, String)>, +) -> Response { + up.record(format!("/flat/{id}/{version}/{file}")).await; + let Some((original, versions)) = up.package(&id) else { + return StatusCode::NOT_FOUND.into_response(); + }; + if !versions.contains(&version) { + return StatusCode::NOT_FOUND.into_response(); + } + if file.ends_with(".nupkg") { + common::make_nupkg(&original, &version).into_response() + } else if file.ends_with(".nuspec") { + format!( + r#" + + + {original} + {version} + test + {original} from upstream + +"# + ) + .into_response() + } else { + StatusCode::NOT_FOUND.into_response() + } +} + +/// Create proxy `name` of `upstream` (no package delay, an hour's metadata +/// TTL) as the bootstrap admin. +async fn create_proxy(srv: &common::TestServer, name: &str, upstream: &Upstream) { + let (u, p) = srv.admin_basic(); + let resp = srv + .client() + .post(format!("{}/admin/repositories", srv.base_url)) + .basic_auth(&u, Some(&p)) + .json(&json!({ + "name": name, + "kind": "proxy", + "remote_url": format!("{}/index.json", upstream.base), + "package_delay_seconds": 0, + "metadata_ttl_seconds": 3600, + })) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201, "create proxy {name}"); +} + +/// `GET` (or `POST`, for a refresh) `path` as the bootstrap admin; the +/// status and the JSON body (`null` when there is none). Fails the test if +/// it takes ten seconds. +async fn admin_send(srv: &common::TestServer, method: reqwest::Method, path: &str) -> (u16, Value) { + let (u, p) = srv.admin_basic(); + let resp = tokio::time::timeout( + Duration::from_secs(10), + srv.client() + .request(method, format!("{}{path}", srv.base_url)) + .basic_auth(&u, Some(&p)) + .send(), + ) + .await + .unwrap_or_else(|_| panic!("{path} took ten seconds")) + .unwrap(); + let status = resp.status().as_u16(); + (status, resp.json().await.unwrap_or(Value::Null)) +} + +/// The dependency ids of `version` on a package detail. +fn dependencies_of(detail: &Value, version: &str) -> Vec { + detail["dependency_groups_by_version"][version] + .as_array() + .into_iter() + .flatten() + .flat_map(|group| { + group["dependencies"] + .as_array() + .cloned() + .unwrap_or_default() + }) + .map(|dep| dep["id"].as_str().unwrap().to_string()) + .collect() +} + +/// The versions of a package detail. +fn versions_of(detail: &Value) -> Vec { + detail["versions"] + .as_array() + .unwrap() + .iter() + .map(|v| v["version"].as_str().unwrap().to_string()) + .collect() +} + +/// Ask `path` (a package page) until its versions are `want`, for ten seconds. +async fn page_until(srv: &common::TestServer, path: &str, want: &[&str]) -> Value { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let (status, detail) = admin_send(srv, reqwest::Method::GET, path).await; + if status == 200 && versions_of(&detail) == want { + return detail; + } + assert!( + tokio::time::Instant::now() < deadline, + "{path} never listed {want:?}: {status} {detail}" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +/// The cached-version rows a proxy holds of a package. +async fn cached_rows(srv: &common::TestServer, proxy: &str, lower: &str) -> i64 { + sqlx::query_scalar( + "SELECT count(*) FROM proxy_cached_version v JOIN repository r ON r.id = v.repository_id + WHERE r.name = $1 AND v.name = $2", + ) + .bind(proxy) + .bind(lower) + .fetch_one(&srv.pool) + .await + .unwrap() +} + +/// A proxy's package page answers from what Forge holds (#836): with the +/// dependencies of the registration a client fetched, and when the upstream +/// listed the package, but without asking the upstream on the request, also +/// once the listing is stale (here: older than a cache invalidation), and +/// without discovering a package it holds nothing of. A stale page starts a +/// refresh off the request, once, and the next look shows what it found. +#[tokio::test] +async fn a_proxy_package_page_answers_from_the_cache_and_refreshes_off_the_request() { + let srv = common::start().await; + let upstream = Upstream::start(&[("Cached.Lib", "1.0.0"), ("Uncached.Lib", "1.0.0")]).await; + let proxy = format!("cachep{}", common::unique()); + create_proxy(&srv, &proxy, &upstream).await; + + // A client restores Cached.Lib: its listing and registration are cached. + for path in [ + format!("/repository/{proxy}/flat/cached.lib/index.json"), + format!("/repository/{proxy}/registration/cached.lib/index.json"), + ] { + let (status, body) = admin_send(&srv, reqwest::Method::GET, &path).await; + assert_eq!(status, 200, "{path}: {body}"); + } + let asked = upstream.requests_for("cached.lib"); + assert!(asked > 0); + + let page = format!("/browse/packages/{proxy}/Cached.Lib"); + let (status, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(status, 200, "{detail}"); + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert_eq!(dependencies_of(&detail, "1.0.0"), ["Dep.Lib"], "{detail}"); + let cached_at = detail["cached_at"].as_str().expect("cached_at").to_string(); + assert!(detail.get("refreshing").is_none(), "fresh: {detail}"); + assert_eq!( + upstream.requests_for("cached.lib"), + asked, + "the page asked the upstream" + ); + + // A version published since shows once the listing is refreshed, not + // before: the page asks nothing within the TTL. + upstream.publish("Cached.Lib", "2.0.0"); + let (_, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert_eq!(upstream.requests_for("cached.lib"), asked); + + // A package the proxy holds nothing of is not found, and looking at its + // page neither asks the upstream nor records it. + let (status, _) = admin_send( + &srv, + reqwest::Method::GET, + &format!("/browse/packages/{proxy}/Uncached.Lib"), + ) + .await; + assert_eq!(status, 404); + assert_eq!(upstream.requests_for("uncached.lib"), 0); + assert_eq!(cached_rows(&srv, &proxy, "uncached.lib").await, 0); + + // Past an invalidation the listing is stale: the page answers from the + // cache and starts a refresh off the request. While the upstream holds + // that refresh's request, the page still answers at once, and starts no + // other. + let (status, _) = admin_send( + &srv, + reqwest::Method::POST, + &format!("/admin/repositories/{proxy}/proxy-cache/invalidate"), + ) + .await; + assert_eq!(status, 200); + upstream.set_open(false); + let seen = upstream.total(); + let (status, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(status, 200, "{detail}"); + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert_eq!(detail["refreshing"], true, "{detail}"); + upstream.asked_beyond(seen).await; + let held = upstream.total(); + let (status, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(status, 200, "{detail}"); + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert_eq!(detail["refreshing"], true, "{detail}"); + assert_eq!(upstream.total(), held, "a second refresh"); + + // Once the upstream answers, the page shows the new version. + upstream.set_open(true); + page_until(&srv, &page, &["1.0.0", "2.0.0"]).await; + // Once the refresh is done (it fetches the registration and tags after + // the listing), the page is fresh again: no other refresh starts. + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let (_, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + if detail.get("refreshing").is_none() { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "still refreshing: {detail}" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + let done = upstream.total(); + let (_, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert!(detail.get("refreshing").is_none(), "{detail}"); + assert!( + detail["cached_at"].as_str().unwrap() > cached_at.as_str(), + "{detail}" + ); + assert_eq!(dependencies_of(&detail, "2.0.0"), ["Dep.Lib"], "{detail}"); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(upstream.total(), done, "a refresh of a fresh page"); +} + +/// **Refresh from upstream** asks the upstream now, and answers the +/// refreshed page: any reader of the proxy may, and it finds a package the +/// proxy held nothing of (#836). A caller who may not read the proxy may not, +/// and a package the upstream does not have stays not found. +#[tokio::test] +async fn refresh_from_upstream_needs_only_read_access() { + let srv = common::start().await; + let upstream = Upstream::start(&[("Fresh.Lib", "1.0.0")]).await; + let proxy = format!("refrp{}", common::unique()); + create_proxy(&srv, &proxy, &upstream).await; + let reader = srv.caller().await; + srv.grant_role(reader.id, &[(proxy.as_str(), "read", None)]) + .await; + let stranger = srv.caller().await; + + let refresh = |caller: &common::Caller, id: &str| { + caller + .apply(srv.client().post(format!( + "{}/browse/packages/{proxy}/{id}/refresh", + srv.base_url + ))) + .send() + }; + let resp = refresh(&stranger, "Fresh.Lib").await.unwrap(); + assert_eq!(resp.status(), 403); + assert_eq!(upstream.requests_for("fresh.lib"), 0); + + let resp = refresh(&reader, "Fresh.Lib").await.unwrap(); + assert_eq!(resp.status(), 200); + let detail: Value = resp.json().await.unwrap(); + assert_eq!(versions_of(&detail), ["1.0.0"], "{detail}"); + assert_eq!(dependencies_of(&detail, "1.0.0"), ["Dep.Lib"], "{detail}"); + assert!(detail["cached_at"].is_string(), "{detail}"); + assert!(upstream.requests_for("fresh.lib") > 0); + + // The page has it now, from the cache. + let resp = reader + .apply(srv.client().get(format!( + "{}/browse/packages/{proxy}/fresh.lib", + srv.base_url + ))) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let resp = refresh(&reader, "Missing.Lib").await.unwrap(); + assert_eq!(resp.status(), 404); +} + +/// A group's package page reads its proxy member from the cache too (#836): +/// a package the proxy holds nothing of is not found through the group and +/// is not asked for, the age of the page is the member's listing, a stale +/// member is refreshed off the request, and Refresh on the group asks the +/// members. +#[tokio::test] +async fn a_group_package_page_reads_its_proxy_members_from_the_cache() { + let srv = common::start().await; + let (u, p) = srv.admin_basic(); + let upstream = Upstream::start(&[("Group.Lib", "1.0.0"), ("Later.Lib", "1.0.0")]).await; + let suffix = common::unique(); + let proxy = format!("cgp{suffix}"); + let group = format!("cgg{suffix}"); + create_proxy(&srv, &proxy, &upstream).await; + let client = srv.client(); + let resp = client + .post(format!("{}/admin/repositories", srv.base_url)) + .basic_auth(&u, Some(&p)) + .json(&json!({"name": group, "kind": "group"})) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201); + let resp = client + .post(format!( + "{}/admin/repositories/{group}/members", + srv.base_url + )) + .basic_auth(&u, Some(&p)) + .json(&json!({"member": proxy, "priority": 10})) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201); + + // A client restores Group.Lib through the group. + let (status, _) = admin_send( + &srv, + reqwest::Method::GET, + &format!("/repository/{group}/flat/group.lib/index.json"), + ) + .await; + assert_eq!(status, 200); + let asked = upstream.requests_for("group.lib"); + + let page = format!("/browse/packages/{group}/Group.Lib"); + let (status, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(status, 200, "{detail}"); + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert!(detail["cached_at"].is_string(), "{detail}"); + assert!(detail.get("refreshing").is_none(), "{detail}"); + assert_eq!(upstream.requests_for("group.lib"), asked); + + let later = format!("/browse/packages/{group}/Later.Lib"); + let (status, _) = admin_send(&srv, reqwest::Method::GET, &later).await; + assert_eq!(status, 404); + assert_eq!(upstream.requests_for("later.lib"), 0); + assert_eq!(cached_rows(&srv, &proxy, "later.lib").await, 0); + + // A stale member is refreshed off the group page's request. + upstream.publish("Group.Lib", "2.0.0"); + let (status, _) = admin_send( + &srv, + reqwest::Method::POST, + &format!("/admin/repositories/{proxy}/proxy-cache/invalidate"), + ) + .await; + assert_eq!(status, 200); + upstream.set_open(false); + let seen = upstream.total(); + let (status, detail) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(status, 200, "{detail}"); + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert_eq!(detail["refreshing"], true, "{detail}"); + upstream.asked_beyond(seen).await; + let (status, _) = admin_send(&srv, reqwest::Method::GET, &page).await; + assert_eq!(status, 200); + upstream.set_open(true); + page_until(&srv, &page, &["1.0.0", "2.0.0"]).await; + + // Refresh on the group asks its members for a package they lacked. + let (status, detail) = + admin_send(&srv, reqwest::Method::POST, &format!("{later}/refresh")).await; + assert_eq!(status, 200, "{detail}"); + assert_eq!(versions_of(&detail), ["1.0.0"]); + assert!(upstream.requests_for("later.lib") > 0); +} diff --git a/docs/changes/2.2.0/836.fixed.md b/docs/changes/2.2.0/836.fixed.md new file mode 100644 index 00000000..8d6f50dd --- /dev/null +++ b/docs/changes/2.2.0/836.fixed.md @@ -0,0 +1,11 @@ +The web UI's package page of a proxy, or of a group with proxy members, asked +the upstream on every view once the package's listing was older than the +proxy's metadata TTL, one step and one member after another: the page was +slow, often showed less than Forge had cached, and opening it added the +package to the browse list. The page now answers from what Forge holds, +without asking the upstream, and says how old that is ("Cached 3 h ago"). +Once that is past the metadata TTL, the page refreshes the package in the +background and shows what it found when it is done. Any reader of the +repository can also refresh the package with **Refresh from upstream**. A +package the proxy holds nothing of shows as not cached, with the same +button, instead of being fetched by looking at it. diff --git a/migrations/0018_proxy_package_listing.sql b/migrations/0018_proxy_package_listing.sql new file mode 100644 index 00000000..ae1c7e72 --- /dev/null +++ b/migrations/0018_proxy_package_listing.sql @@ -0,0 +1,23 @@ +-- --------------------------------------------------------------------------- +-- proxy_package_listing: when a proxy's upstream last listed the versions of +-- a package (#836) +-- +-- The web UI's package page answers from what Forge holds, without asking +-- the upstream on the request: it says how old that is ("cached 3 h ago"), +-- and starts a refresh off the request once the listing is past the +-- repository's metadata TTL, or older than the repository's last edit or +-- cache invalidation (`repository.updated_at`). The "listing refreshed" +-- marker the read path keeps for the TTL lives in the cache, which may be +-- per instance and forgets on restart, so the time is kept here. +-- +-- One row per repository and package id as the proxy stores it (lowercased), +-- written each time the upstream answers the package's version listing. +-- Repointing the proxy to another upstream deletes its rows; deleting the +-- repository deletes them with it. +-- --------------------------------------------------------------------------- +CREATE TABLE proxy_package_listing ( + repository_id bigint NOT NULL REFERENCES repository(id) ON DELETE CASCADE, + name text NOT NULL, + listed_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (repository_id, name) +); diff --git a/web/src/api/packages.ts b/web/src/api/packages.ts index 77bedd0a..5bf7dc13 100644 --- a/web/src/api/packages.ts +++ b/web/src/api/packages.ts @@ -17,6 +17,12 @@ import { ApiError, httpClient } from '~/lib/httpClient' * `_app` layout's login gate ensures these are only called with a session. * An offline hosted or group repository answers its package detail and its * per-repository search with 503 (see {@link isRepositoryOffline}). + * + * A proxy's and a group's package detail is what Forge has cached: the + * request asks no upstream, says when the upstream last listed the package + * (`cached_at`), and refreshes a stale listing in the background + * (`refreshing`). A package the proxy holds nothing of is a 404 until a + * client installs it or a reader refreshes it ({@link refreshPackage}). */ /** One readable repository (`GET /browse/repos` item) — a folder-view root. */ @@ -186,6 +192,27 @@ export interface PackageDetail { * proxies of formats without one. */ package_delay?: AppliedDelay + /** + * When the upstream last listed the package's versions: a proxy's, or the + * oldest among the proxies a group serves it from. The page shows what + * Forge held then, and what clients fetched since. Absent for a hosted + * repository and when no time was recorded. + */ + cached_at?: string + /** + * A refresh from the upstream is under way in the background: looking + * again shortly shows what it found. + */ + refreshing?: boolean +} + +/** + * Whether a repository of `kind` shows what Forge has cached of an upstream + * (a proxy, or a group, through its proxies): its package page can be + * refreshed from the upstream. + */ +export function showsCache(kind: string | undefined): boolean { + return kind === 'proxy' || kind === 'group' } /** Search parameters for the browse list (mirrors `BrowseQuery`). */ @@ -270,6 +297,17 @@ export function fetchPackageDetail( ) } +/** + * **Refresh from upstream**: ask the upstream for the package now (a group: + * its proxies' upstreams), as a client's install does, and get the refreshed + * detail. Any reader may. A 404 means the upstream does not have it either. + */ +export function refreshPackage(repo: string, id: string): Promise { + return httpClient.post( + `/browse/packages/${encodeURIComponent(repo)}/${encodeURIComponent(id)}/refresh`, + ) +} + /** Hard-delete a hosted package version (admin only). */ export function deletePackageVersion( repo: string, diff --git a/web/src/lib/age.test.ts b/web/src/lib/age.test.ts new file mode 100644 index 00000000..2146630a --- /dev/null +++ b/web/src/lib/age.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, it } from 'vitest' + +import { formatAge } from './age' + +describe('formatAge', () => { + const now = Date.parse('2026-11-03T12:00:00Z') + + it('says how long ago a time was, in the largest unit that fits', () => { + expect(formatAge('2026-11-03T11:59:30Z', now)).toBe('just now') + expect(formatAge('2026-11-03T11:55:00Z', now)).toBe('5 min ago') + expect(formatAge('2026-11-03T09:00:00Z', now)).toBe('3 h ago') + expect(formatAge('2026-11-02T00:00:00Z', now)).toBe('36 h ago') + expect(formatAge('2026-10-31T12:00:00Z', now)).toBe('3 d ago') + }) + + it('takes a time ahead of the clock for now, and refuses what does not parse', () => { + expect(formatAge('2026-11-03T12:05:00Z', now)).toBe('just now') + expect(formatAge('yesterday', now)).toBeNull() + }) +}) diff --git a/web/src/lib/age.ts b/web/src/lib/age.ts new file mode 100644 index 00000000..a61a13e8 --- /dev/null +++ b/web/src/lib/age.ts @@ -0,0 +1,16 @@ +/** + * How long ago `iso` was, the way a page says how old its data is: "just + * now", "5 min ago", "3 h ago", "2 d ago". `null` when `iso` does not parse. + * A time in the future (a clock ahead of this one) is "just now". + */ +export function formatAge(iso: string, now: number = Date.now()): string | null { + const at = new Date(iso).getTime() + if (Number.isNaN(at)) return null + const seconds = Math.max(0, Math.floor((now - at) / 1000)) + if (seconds < 60) return 'just now' + const minutes = Math.floor(seconds / 60) + if (minutes < 60) return `${minutes} min ago` + const hours = Math.floor(minutes / 60) + if (hours < 48) return `${hours} h ago` + return `${Math.floor(hours / 24)} d ago` +} diff --git a/web/src/queries/usePackageDetail.ts b/web/src/queries/usePackageDetail.ts index 3082594d..1d8b2641 100644 --- a/web/src/queries/usePackageDetail.ts +++ b/web/src/queries/usePackageDetail.ts @@ -5,18 +5,41 @@ import { fetchPackageDetail, deletePackageVersion, pkgKeys, + refreshPackage, type PackageDetail, } from '~/api/packages' +/** How often a page whose refresh runs in the background looks again. */ +export const REFRESH_POLL_MS = 2000 + /** * Package detail (`GET /browse/packages/{repo}/{id}`): visible versions with * per-version metadata + dependency groups. A 404 (package not found) surfaces - * as an `ApiError(404)` the route renders as an empty/not-found state. + * as an `ApiError(404)` the route renders as an empty/not-found state. While + * the server refreshes the package from its upstream in the background + * (`refreshing`), it is asked again every {@link REFRESH_POLL_MS}. */ export function usePackageDetail(repo: string, id: string) { return useQuery({ queryKey: pkgKeys.detail(repo, id), queryFn: ({ signal }) => fetchPackageDetail(repo, id, signal), + refetchInterval: (query) => (query.state.data?.refreshing ? REFRESH_POLL_MS : false), + }) +} + +/** + * **Refresh from upstream** (`POST /browse/packages/{repo}/{id}/refresh`): + * the refreshed detail replaces the page's, and the repository's package + * list is read again, which may now have the package. + */ +export function useRefreshPackage(repo: string, id: string) { + const queryClient = useQueryClient() + return useMutation({ + mutationFn: () => refreshPackage(repo, id), + onSuccess: (detail) => { + queryClient.setQueryData(pkgKeys.detail(repo, id), detail) + void queryClient.invalidateQueries({ queryKey: [...pkgKeys.all(), 'search'] }) + }, }) } diff --git a/web/src/routes/PackageDetail.test.tsx b/web/src/routes/PackageDetail.test.tsx index 90d7e13c..7802c188 100644 --- a/web/src/routes/PackageDetail.test.tsx +++ b/web/src/routes/PackageDetail.test.tsx @@ -22,7 +22,9 @@ import { searchOptions } from '~/lib/search' * a reason; a proxy labels the time column "Cached" (when * Forge downloaded it), a hosted repository "Published"; a 404 renders the * not-found empty state; an offline repository's 503 says so (without - * retrying). Drives the real route tree + `usePackageDetail`. + * retrying); a proxy's page says how old its cache is, follows a background + * refresh, and is refreshed from the upstream on request, also when the + * package is not cached yet. Drives the real route tree + `usePackageDetail`. */ const anonMe: Me = { @@ -116,6 +118,7 @@ function mockApi( format = 'nuget', deleteVersion, release, + refresh, repos = [{ name: 'internal', format }], }: { me?: Me @@ -123,7 +126,9 @@ function mockApi( deleteVersion?: () => Response | Promise /** `POST …/release-delay`: the path and the body. */ release?: (path: string, body: unknown) => Response - repos?: { name: string; format: string }[] + /** `POST /browse/packages/{repo}/{id}/refresh`. */ + refresh?: (repo: string, id: string) => Response + repos?: { name: string; format: string; kind?: string; online?: boolean }[] } = {}, ) { fetchMock.mockImplementation((input: RequestInfo | URL, init?: RequestInit) => { @@ -148,6 +153,10 @@ function mockApi( if (init?.method === 'POST' && release && url.pathname.endsWith('/release-delay')) { return Promise.resolve(release(url.pathname, JSON.parse(init.body as string))) } + const refreshed = url.pathname.match(/\/browse\/packages\/([^/]+)\/([^/]+)\/refresh$/) + if (init?.method === 'POST' && refresh && refreshed) { + return Promise.resolve(refresh(refreshed[1], refreshed[2])) + } if (url.pathname.endsWith('/browse/packages')) { return Promise.resolve(jsonResponse({ total: 0, items: [] })) } @@ -496,6 +505,89 @@ describe('PackageDetail', () => { expect(screen.queryByRole('columnheader', { name: 'Cached' })).toBeNull() }) + it("says how old a proxy's page is and refreshes it from the upstream", async () => { + const cachedAt = new Date(Date.now() - 3 * 3600_000).toISOString() + const newer: PackageDetail = { + ...proxyDetail, + versions: [ + ...proxyDetail.versions, + { ...proxyDetail.versions[0], version: '13.0.4', version_original: '13.0.4' }, + ], + cached_at: new Date().toISOString(), + } + const refresh = vi.fn(() => jsonResponse(newer)) + mockApi(() => jsonResponse({ ...proxyDetail, cached_at: cachedAt }), { + repos: [{ name: 'proxy', format: 'nuget', kind: 'proxy' }], + refresh, + }) + + await renderDetail('/packages/proxy/newtonsoft.json') + + await screen.findByRole('heading', { name: 'Newtonsoft.Json' }) + expect(screen.getByText('Cached 3 h ago')).toBeInTheDocument() + expect(screen.queryByRole('cell', { name: '13.0.4' })).toBeNull() + await userEvent.click(screen.getByRole('button', { name: 'Refresh from upstream' })) + + expect(await screen.findByText('Cached just now')).toBeInTheDocument() + expect(refresh).toHaveBeenCalledWith('proxy', 'newtonsoft.json') + expect(screen.getAllByText('13.0.4').length).toBeGreaterThan(0) + }) + + it('looks again while the server refreshes the package in the background', async () => { + let calls = 0 + mockApi( + () => { + calls += 1 + return jsonResponse( + calls === 1 + ? { ...proxyDetail, cached_at: '2026-01-01T00:00:00Z', refreshing: true } + : { ...proxyDetail, cached_at: new Date().toISOString() }, + ) + }, + { repos: [{ name: 'proxy', format: 'nuget', kind: 'proxy' }] }, + ) + + await renderDetail('/packages/proxy/newtonsoft.json') + + expect(await screen.findByText(/refreshing from the upstream/)).toBeInTheDocument() + expect(await screen.findByText('Cached just now', {}, { timeout: 5000 })).toBeInTheDocument() + expect(screen.queryByText(/refreshing from the upstream/)).toBeNull() + }) + + it("offers to refresh a package a proxy has not cached, and says when the upstream lacks it", async () => { + const refresh = vi + .fn<(repo: string, id: string) => Response>() + .mockReturnValueOnce(new Response('package not found', { status: 404 })) + .mockReturnValueOnce(jsonResponse(proxyDetail)) + mockApi(() => new Response('package not found', { status: 404 }), { + repos: [{ name: 'proxy', format: 'nuget', kind: 'proxy' }], + refresh, + }) + + await renderDetail('/packages/proxy/newtonsoft.json') + + expect(await screen.findByText('Package not cached')).toBeInTheDocument() + expect(screen.queryByText('Package not found')).toBeNull() + const button = screen.getByRole('button', { name: 'Refresh from upstream' }) + await userEvent.click(button) + expect(await screen.findByText('Not found upstream either')).toBeInTheDocument() + + await userEvent.click(screen.getByRole('button', { name: 'Refresh from upstream' })) + expect(await screen.findByRole('heading', { name: 'Newtonsoft.Json' })).toBeInTheDocument() + expect(refresh).toHaveBeenCalledTimes(2) + }) + + it('offers no refresh on a hosted repository, nor on an offline proxy', async () => { + mockApi(() => new Response('package not found', { status: 404 }), { + repos: [{ name: 'proxy', format: 'nuget', kind: 'proxy', online: false }], + }) + + await renderDetail('/packages/proxy/missing') + + expect(await screen.findByText('Package not found')).toBeInTheDocument() + expect(screen.queryByRole('button', { name: 'Refresh from upstream' })).toBeNull() + }) + it('renders a not-found empty state on 404', async () => { mockApi(() => new Response('package not found', { status: 404 })) diff --git a/web/src/routes/PackageDetail.tsx b/web/src/routes/PackageDetail.tsx index 523f41d8..0f4619af 100644 --- a/web/src/routes/PackageDetail.tsx +++ b/web/src/routes/PackageDetail.tsx @@ -5,11 +5,19 @@ import { ApiError } from '~/lib/httpClient' import { delaySourceLabel, isRepositoryOffline, + showsCache, type AppliedDelay, type DetailVersion, + type PackageDetail, } from '~/api/packages' +import { formatAge } from '~/lib/age' +import { formatDateTime } from '~/lib/oci' import { useBrowseRepos } from '~/queries/useBrowseRepos' -import { useDeletePackageVersion, usePackageDetail } from '~/queries/usePackageDetail' +import { + useDeletePackageVersion, + usePackageDetail, + useRefreshPackage, +} from '~/queries/usePackageDetail' import { useMe } from '~/queries/useMe' import { useReleaseFromDelay } from '~/queries/useRepositories' import { VersionTable } from '~/components/VersionTable' @@ -18,6 +26,7 @@ import { DependencyGroups } from '~/components/DependencyGroups' import { ErrorState, RepositoryOfflineState } from '~/components/ErrorState' import { EmptyState } from '~/components/EmptyState' import { + Button, Code, CodeBlock, Columns, @@ -49,6 +58,10 @@ function PackageDetailRoute() { * dist-tags), deprecations and README when its format has them. Loading / * empty / not-found / repository-offline / error states are all handled. * Reachable by any reader. + * + * A proxy's or group's page is what Forge has cached: it says how old that + * is, follows a refresh the server runs in the background, and offers + * **Refresh from upstream**, also for a package not cached yet. */ function PackageDetailPage() { const { repo, id } = Route.useParams() @@ -61,11 +74,21 @@ function PackageDetailPage() { const [pendingRelease, setPendingRelease] = useState(null) const [reason, setReason] = useState('') const toast = useToast() - // The repository's format, from the (cached) list the rail shows too. + const refresh = useRefreshPackage(repo, id) + // The repository's format and kind, from the (cached) list the rail shows too. const repos = useBrowseRepos() - const format = Array.isArray(repos.data) - ? repos.data.find((r) => r.name === repo)?.format + const repoInfo = Array.isArray(repos.data) + ? repos.data.find((r) => r.name === repo) : undefined + const format = repoInfo?.format + // Refreshing asks an upstream: an offline proxy asks none. + const refreshable = + showsCache(detail.data?.repo_kind ?? repoInfo?.kind) && repoInfo?.online !== false + const askRefresh = () => + refresh.mutate(undefined, { + onSuccess: (refreshed) => + toast({ tone: 'ok', title: `Refreshed ${refreshed.original_id} from the upstream` }), + }) const crumbs = [ { label: 'Repositories', to: '/' }, { label: repo, to: '/repos/$repo', params: { repo } }, @@ -88,7 +111,16 @@ function PackageDetailPage() { return ( - {notFound ? ( + {notFound && refreshable ? ( + + ) : notFound ? ( No readable package {id} in repository {repo}. @@ -127,8 +159,28 @@ function PackageDetailPage() { meta={[ pkg.versions.length === 1 ? '1 version' : `${pkg.versions.length} versions`, ...(latest ? [`latest ${latest}`] : []), + cacheAge(pkg), ]} + actions={ + refreshable && ( + + ) + } /> + {refresh.isError && ( + + )} @@ -346,6 +398,85 @@ function PackageDetailPage() { ) } +/** + * How old a proxy's or group's page is ("Cached 3 h ago", the time itself + * on hover), and whether a refresh runs in the background; nothing for a + * hosted repository. + */ +function cacheAge(pkg: PackageDetail) { + const age = pkg.cached_at ? formatAge(pkg.cached_at) : null + if (!age && !pkg.refreshing) return null + return ( + + {age && `Cached ${age}`} + {age && pkg.refreshing && ' · '} + {pkg.refreshing && 'refreshing from the upstream…'} + + ) +} + +/** + * A proxy's or group's 404: Forge holds nothing of the package there (the + * page shows only what is cached), with **Refresh from upstream**. A refresh + * that finds nothing either says so. + */ +function NotCached({ + repo, + id, + group, + onRefresh, + pending, + error, +}: { + repo: string + id: string + group: boolean + onRefresh: () => void + pending: boolean + error: ApiError | null +}) { + return ( + <> + + {pending ? 'Refreshing…' : 'Refresh from upstream'} + + } + > + {group ? ( + <> + No member of {repo} holds {id}. + + ) : ( + <> + {repo} has not cached {id}. + + )}{' '} + Browsing shows what Forge has cached; a package is cached once a client installs it. + Refresh asks the upstream for it now. + + {error?.status === 404 ? ( + + {group ? "The group's upstreams do not have " : 'The upstream does not have '} + {id}, or this account may not read it. + + ) : ( + error && ( + + ) + )} + + ) +} + /** "1 h (repository default)", "none (trusted-vendors)", "14 d (risky.*)". */ function describeDelay(delay: AppliedDelay): string { const length = delay.delay_seconds > 0 ? formatDelay(delay.delay_seconds) : 'none'