diff --git a/crates/peryx-ecosystem-oci/src/quota.rs b/crates/peryx-ecosystem-oci/src/quota.rs index bbb29fd2..92ec4a2f 100644 --- a/crates/peryx-ecosystem-oci/src/quota.rs +++ b/crates/peryx-ecosystem-oci/src/quota.rs @@ -1,4 +1,25 @@ -use peryx_storage::meta::{AccountingClass, NewQuotaReservation}; +//! Repository-quota enforcement for the hosted push paths. +//! +//! A blob upload, cross-repo mount, or manifest publication reserves capacity against the substrate +//! before it becomes discoverable, commits that reservation atomically with the metadata write, and +//! releases it when the write fails. An index that configures no quota keeps its original write path, +//! so an unmetered registry pays nothing for the machinery. + +use axum::response::Response; +use peryx_core::Role; +use peryx_driver::ServingState; +use peryx_events::metrics::{Event, MetricFamily}; +use peryx_index::Index; +use peryx_policy::Policy; +use peryx_storage::meta::{ + AccountingClass, DriverTxn, MetaStore, NewQuotaReservation, QuotaError, QuotaLimit, QuotaLimits, + QuotaReservationRecord, +}; + +use crate::error::{ErrorCode, error_response}; +use crate::name::Reference; +use crate::registry::ServeError; +use crate::store::{self, Manifest}; /// Account the OCI repository path as a project and an optional tag as its version. #[must_use] @@ -21,3 +42,253 @@ pub const fn quota_reservation<'a>( created_at_unix, } } + +/// A hosted push admitted against the repository quota. +const QUOTA_ADMITTED_FAMILY: MetricFamily = MetricFamily { + key: "quota_admitted", + prom_name: "peryx_oci_quota_admitted_total", + help: "Hosted OCI pushes admitted against the repository quota.", + ui_label: "Quota admitted pushes", + roles: &[Role::Hosted], +}; + +/// A hosted push refused by the repository quota. +const QUOTA_REJECTED_FAMILY: MetricFamily = MetricFamily { + key: "quota_rejected", + prom_name: "peryx_oci_quota_rejected_total", + help: "Hosted OCI pushes refused by the repository quota.", + ui_label: "Quota rejected pushes", + roles: &[Role::Hosted], +}; + +/// The quota-decision counters the OCI driver publishes. +pub const QUOTA_FAMILIES: &[MetricFamily] = &[QUOTA_ADMITTED_FAMILY, QUOTA_REJECTED_FAMILY]; + +/// The outcome of admitting a hosted push against the repository quota. +pub enum Admission { + /// The index configures no quota; publish without accounting. + Unmetered, + /// The push is admitted. Commit the reservation with the publication, or release it on failure. + Reserved(QuotaReservationRecord), + /// The push is refused. Return this distribution-spec error to the client. + Rejected(Response), +} + +/// Reserve repository capacity for a hosted push and record the decision metric. +/// +/// Returns [`Admission::Unmetered`] when the index sets no quota, so an unconfigured registry keeps +/// its original write path. In audit mode the reservation is admitted even when it crosses a limit, +/// and its recorded violations stay on the durable reservation record for inspection. +pub fn admit_push( + state: &ServingState, + index: &Index, + repo: &str, + version: Option<&str>, + digest: &str, + bytes: u64, +) -> Result { + let Some(limits) = quota_limits(&index.policy) else { + return Ok(Admission::Unmetered); + }; + let request = quota_reservation( + &index.name, + repo, + version, + digest, + bytes, + AccountingClass::Hosted, + (state.clock)(), + ); + match reserve(&state.meta, request, limits)? { + ReserveOutcome::Admitted(record) => { + record_quota_metric(state, index, repo, QUOTA_ADMITTED_FAMILY.key); + Ok(Admission::Reserved(record)) + } + ReserveOutcome::Rejected(violations) => { + record_quota_metric(state, index, repo, QUOTA_REJECTED_FAMILY.key); + Ok(Admission::Rejected(error_response( + ErrorCode::Denied, + &format!("repository quota exceeded: {}", describe(&violations)), + ))) + } + } +} + +/// The storage limit set an index configures, or `None` when it accounts for nothing. The per-file +/// size limit is enforced on the byte stream itself, so it alone does not switch accounting on. +fn quota_limits(policy: &Policy) -> Option { + policy.enforces_quota().then(|| QuotaLimits { + max_file_bytes: policy.max_file_size(), + max_accounted_bytes: policy.max_accounted_bytes(), + max_projects: policy.max_projects(), + max_versions_per_project: policy.max_versions_per_project(), + audit: policy.quota_audit(), + }) +} + +/// The reservation decision, separated from request state so the enforce, audit, and fault branches +/// are exercised against a bare [`MetaStore`]. +enum ReserveOutcome { + Admitted(QuotaReservationRecord), + Rejected(Vec), +} + +fn reserve( + meta: &MetaStore, + request: NewQuotaReservation<'_>, + limits: QuotaLimits, +) -> Result { + match meta.reserve_quota(request, limits) { + Ok(record) => Ok(ReserveOutcome::Admitted(record)), + Err(QuotaError::Exceeded { violations }) => Ok(ReserveOutcome::Rejected(violations)), + Err(err) => Err(err.into()), + } +} + +fn record_quota_metric(state: &ServingState, index: &Index, repo: &str, family: &'static str) { + state.metrics.record(Event::Ecosystem { + route: index.route.clone(), + project: repo.to_owned(), + filename: None, + family, + }); +} + +fn describe(violations: &[QuotaLimit]) -> String { + violations + .iter() + .map(|limit| match limit { + QuotaLimit::FileBytes => "file size", + QuotaLimit::AccountedBytes => "repository bytes", + QuotaLimit::Projects => "repository projects", + QuotaLimit::VersionsPerProject => "project versions", + }) + .collect::>() + .join(", ") +} + +/// Publish a blob's `(index, repo)` membership, committing a quota reservation with it when the push +/// was metered so the two land in one transaction. +pub fn commit_blob_membership( + meta: &MetaStore, + index: &str, + repo: &str, + digest: &str, + reservation: Option, +) -> Result<(), ServeError> { + match reservation { + None => Ok(store::record_blob_membership(meta, index, repo, digest)?), + Some(record) => meta.commit_driver_txn_with_quota(record.id, |txn| { + txn.put(&store::blob_membership_key(index, repo, digest), &[])?; + Ok::<_, ServeError>(((), Vec::new())) + }), + } +} + +/// Publish a manifest by digest and optional tag, committing a quota reservation with it when the +/// push was metered. Reports whether the searchable tag set grew. +pub fn publish_manifest( + meta: &MetaStore, + index: &str, + repo: &str, + canonical: &str, + manifest: &Manifest, + reference: &Reference, + reservation: Option, +) -> Result { + let body = |txn: &mut DriverTxn| -> Result<(bool, Vec>), ServeError> { + store::record_manifest_txn(txn, index, repo, canonical, manifest)?; + let grew = match reference { + Reference::Tag(tag) => store::put_tag_txn(txn, index, repo, tag, canonical)?, + Reference::Digest(_) => false, + }; + Ok((grew, Vec::new())) + }; + match reservation { + None => meta.commit_driver_txn(body), + Some(record) => meta.commit_driver_txn_with_quota(record.id, body), + } +} + +/// Whether this exact manifest is already published under `reference`, so a re-push is a no-op that +/// must not account a fresh version or byte allocation. +pub fn manifest_already_published( + meta: &MetaStore, + index: &str, + repo: &str, + canonical: &str, + reference: &Reference, +) -> Result { + if !store::manifest_is_member(meta, index, repo, canonical)? { + return Ok(false); + } + match reference { + Reference::Digest(_) => Ok(true), + Reference::Tag(tag) => Ok(store::get_tag(meta, index, repo, tag)?.as_deref() == Some(canonical)), + } +} + +#[cfg(test)] +mod tests { + use peryx_storage::meta::MetaStore; + + use super::{ReserveOutcome, describe, quota_reservation, reserve}; + use peryx_storage::meta::{AccountingClass, QuotaLimit, QuotaLimits}; + + fn store() -> (tempfile::TempDir, MetaStore) { + let dir = tempfile::tempdir().unwrap(); + let meta = MetaStore::open(dir.path().join("peryx.redb")).unwrap(); + (dir, meta) + } + + #[test] + fn test_reserve_admits_within_the_limit() { + let (_dir, meta) = store(); + let request = quota_reservation("store", "app", None, "sha256:a", 4, AccountingClass::Hosted, 1); + let limits = QuotaLimits { + max_accounted_bytes: Some(8), + ..QuotaLimits::default() + }; + assert!(matches!( + reserve(&meta, request, limits).unwrap(), + ReserveOutcome::Admitted(_) + )); + } + + #[test] + fn test_reserve_rejects_over_the_limit_in_enforce_mode() { + let (_dir, meta) = store(); + let request = quota_reservation("store", "app", None, "sha256:a", 9, AccountingClass::Hosted, 1); + let limits = QuotaLimits { + max_accounted_bytes: Some(8), + ..QuotaLimits::default() + }; + assert!(matches!( + reserve(&meta, request, limits).unwrap(), + ReserveOutcome::Rejected(violations) if violations == vec![QuotaLimit::AccountedBytes] + )); + } + + #[test] + fn test_reserve_maps_a_validation_fault_to_a_serve_error() { + let (_dir, meta) = store(); + // A repository key past the substrate's identity length ceiling is a hard fault, not a quota + // decision, so it propagates rather than reads as an admission or a rejection. + let long = "r".repeat(600); + let request = quota_reservation(&long, "app", None, "sha256:a", 1, AccountingClass::Hosted, 1); + assert!(reserve(&meta, request, QuotaLimits::default()).is_err()); + } + + #[test] + fn test_describe_names_each_crossed_counter() { + assert_eq!( + describe(&[ + QuotaLimit::FileBytes, + QuotaLimit::AccountedBytes, + QuotaLimit::Projects, + QuotaLimit::VersionsPerProject, + ]), + "file size, repository bytes, repository projects, project versions" + ); + } +} diff --git a/crates/peryx-ecosystem-oci/src/registry/blobs/mod.rs b/crates/peryx-ecosystem-oci/src/registry/blobs/mod.rs index efd11192..a3a268f5 100644 --- a/crates/peryx-ecosystem-oci/src/registry/blobs/mod.rs +++ b/crates/peryx-ecosystem-oci/src/registry/blobs/mod.rs @@ -441,10 +441,11 @@ fn download_error_response(err: DownloadError) -> Response { pub(super) async fn commit_blob( state: &ServingState, pending: BlobWrite, - index: &str, + index: &Index, repo: &str, name: &str, digest: &str, + bytes: u64, ) -> Result { let Some(storage) = store::blob_digest(digest) else { return Ok(error_response( @@ -452,12 +453,30 @@ pub(super) async fn commit_blob( "only sha256 blob digests are supported", )); }; + // A digest this repository already serves is accounted; re-pushing it must not reserve again. + let reservation = if store::blob_is_member(&state.meta, &index.name, repo, digest)? { + None + } else { + match crate::quota::admit_push(state, index, repo, None, digest, bytes)? { + crate::quota::Admission::Rejected(response) => { + pending.abort().await.map_err(blob_fault)?; + return Ok(response); + } + crate::quota::Admission::Unmetered => None, + crate::quota::Admission::Reserved(record) => Some(record), + } + }; match pending.commit(&storage).await { Ok(()) => { - store::record_blob_membership(&state.meta, index, repo, digest)?; + crate::quota::commit_blob_membership(&state.meta, &index.name, repo, digest, reservation)?; Ok(blob_created(name, digest)) } - Err(err) => Ok(download_error_response(DownloadError::Blob(err))), + Err(err) => { + if let Some(record) = reservation { + state.meta.release_quota_reservation(record.id)?; + } + Ok(download_error_response(DownloadError::Blob(err))) + } } } diff --git a/crates/peryx-ecosystem-oci/src/registry/manifests/write.rs b/crates/peryx-ecosystem-oci/src/registry/manifests/write.rs index fedb69e5..ada39fc1 100644 --- a/crates/peryx-ecosystem-oci/src/registry/manifests/write.rs +++ b/crates/peryx-ecosystem-oci/src/registry/manifests/write.rs @@ -72,10 +72,31 @@ pub(in crate::registry) async fn put_manifest( media_type: media_type.clone(), bytes: bytes.to_vec(), }; - store::record_manifest(&state.meta, &index.name, &repo, &canonical, &manifest)?; - if let Reference::Tag(tag) = reference - && store::put_tag(&state.meta, &index.name, &repo, tag, &canonical)? - { + let version = match reference { + Reference::Tag(tag) => Some(tag.as_str()), + Reference::Digest(_) => None, + }; + // A re-push of the same manifest under the same reference is already accounted, so it must not + // reserve a fresh version or byte allocation. + let reservation = + if crate::quota::manifest_already_published(&state.meta, &index.name, &repo, &canonical, reference)? { + None + } else { + match crate::quota::admit_push(state, index, &repo, version, &canonical, bytes.len() as u64)? { + crate::quota::Admission::Rejected(response) => return Ok(response), + crate::quota::Admission::Unmetered => None, + crate::quota::Admission::Reserved(record) => Some(record), + } + }; + if crate::quota::publish_manifest( + &state.meta, + &index.name, + &repo, + &canonical, + &manifest, + reference, + reservation, + )? { state.bump_search_epoch(); } let subject = record_referrer(state, &index.name, &repo, &canonical, &media_type, &bytes)?; diff --git a/crates/peryx-ecosystem-oci/src/registry/mod.rs b/crates/peryx-ecosystem-oci/src/registry/mod.rs index 1dc1ca63..3f9b0d9a 100644 --- a/crates/peryx-ecosystem-oci/src/registry/mod.rs +++ b/crates/peryx-ecosystem-oci/src/registry/mod.rs @@ -95,6 +95,11 @@ impl From for ServeError { Self::Transport(err.to_string()) } } +impl From for ServeError { + fn from(err: peryx_storage::meta::QuotaError) -> Self { + Self::Transport(err.to_string()) + } +} impl ServeError { /// A user-visible one-line description of the fault, for the web view methods that surface a /// message rather than a `/v2/` response. @@ -222,6 +227,10 @@ impl EcosystemDriver for OciRe peryx_core::Ecosystem::Oci } + fn metric_families(&self) -> &'static [peryx_events::metrics::MetricFamily] { + crate::quota::QUOTA_FAMILIES + } + fn mount(&self) -> RouteMount { RouteMount::Absolute(&["/v2/"]) } diff --git a/crates/peryx-ecosystem-oci/src/registry/uploads.rs b/crates/peryx-ecosystem-oci/src/registry/uploads.rs index 870c9337..8ceb2205 100644 --- a/crates/peryx-ecosystem-oci/src/registry/uploads.rs +++ b/crates/peryx-ecosystem-oci/src/registry/uploads.rs @@ -33,13 +33,25 @@ impl OciRegistryWithHasher } if let Some((source_index, source_repo)) = resolve(&state.indexes, source) && !policy_blocks(source_index, PolicyAction::Serve, source_repo) - && state.blobs.head(&storage).await.map_err(blob_fault)?.is_some() + && let Some(metadata) = state.blobs.head(&storage).await.map_err(blob_fault)? && self.blob_authorized(state, source_index, source_repo, mount)? { if policy_blocks(index, PolicyAction::Upload, &repo) { return Ok(error_response(ErrorCode::Denied, "image name is blocked by policy")); } - store::record_blob_membership(&state.meta, &index.name, &repo, mount)?; + // A mount publishes an existing blob into this repository without a transfer, so it + // reserves the mounted digest's bytes exactly as an upload of them would; a digest + // already served here is not reserved again. + let reservation = if store::blob_is_member(&state.meta, &index.name, &repo, mount)? { + None + } else { + match crate::quota::admit_push(state, index, &repo, None, mount, metadata.bytes)? { + crate::quota::Admission::Rejected(response) => return Ok(response), + crate::quota::Admission::Unmetered => None, + crate::quota::Admission::Reserved(record) => Some(record), + } + }; + crate::quota::commit_blob_membership(&state.meta, &index.name, &repo, mount, reservation)?; return Ok(blob_created(name, mount)); } } @@ -49,7 +61,7 @@ impl OciRegistryWithHasher if let Err(err) = append_body(&mut pending, &mut size, body, index, &repo).await { return err.into_response(); } - return commit_blob(state, pending, &index.name, &repo, name, digest).await; + return commit_blob(state, pending, index, &repo, name, digest, size).await; } let now = (state.clock)(); let session = Self::random_session()?; @@ -228,7 +240,7 @@ impl OciRegistryWithHasher "finishing an upload requires a digest", )); }; - commit_blob(state, entry.pending, &index.name, &repo, name, &digest).await + commit_blob(state, entry.pending, index, &repo, name, &digest, entry.offset).await } } diff --git a/crates/peryx-ecosystem-oci/src/store.rs b/crates/peryx-ecosystem-oci/src/store.rs index 7039b8c3..d17811ce 100644 --- a/crates/peryx-ecosystem-oci/src/store.rs +++ b/crates/peryx-ecosystem-oci/src/store.rs @@ -8,7 +8,7 @@ use std::collections::BTreeSet; -use peryx_storage::meta::{MetaError, MetaStore}; +use peryx_storage::meta::{DriverTxn, MetaError, MetaStore}; /// The driver-KV prefix every manifest is keyed under, its digest following. mod descriptors; @@ -66,10 +66,13 @@ fn tag_prefix(index: &str, repo: &str) -> String { format!("{TAG_PREFIX}{index}\u{0}{repo}\u{0}") } -/// Store a manifest under its digest. +/// Store a manifest under its digest, without recording membership — a test seed for the by-digest +/// read and delete paths that expects a manifest present in the global pool but served by no +/// repository. /// /// # Errors /// Returns a store error if the write fails. +#[cfg(test)] pub fn put_manifest(meta: &MetaStore, digest: &str, manifest: &Manifest) -> Result<(), MetaError> { meta.put_driver_value(&manifest_key(digest), &manifest.encode()) } @@ -88,25 +91,38 @@ pub fn record_manifest( digest: &str, manifest: &Manifest, ) -> Result<(), MetaError> { - put_manifest(meta, digest, manifest)?; - record_membership(meta, index, repo, digest)?; + meta.commit_driver_txn(|txn| { + record_manifest_txn(txn, index, repo, digest, manifest)?; + Ok(((), Vec::new())) + }) +} + +/// Stage a manifest and its `(index, repo)` memberships inside an open transaction, so a caller can +/// publish it atomically with a quota-reservation commit. +/// +/// # Errors +/// Returns a store error if a write fails. +pub fn record_manifest_txn( + txn: &mut DriverTxn, + index: &str, + repo: &str, + digest: &str, + manifest: &Manifest, +) -> Result<(), MetaError> { + txn.put(&manifest_key(digest), &manifest.encode())?; + txn.put(&membership_key(index, repo, digest), &[])?; let (children, blobs) = manifest_descriptors(&manifest.bytes); for child in children { - record_membership(meta, index, repo, &child)?; + txn.put(&membership_key(index, repo, &child), &[])?; } for blob in blobs { - record_blob_membership(meta, index, repo, &blob)?; + txn.put(&blob_membership_key(index, repo, &blob), &[])?; } Ok(()) } -/// Record that `(index, repo)` serves `digest`. The value is empty: the key's presence is the -/// authorization a by-digest read checks. -fn record_membership(meta: &MetaStore, index: &str, repo: &str, digest: &str) -> Result<(), MetaError> { - meta.put_driver_value(&membership_key(index, repo, digest), &[]) -} - -/// The driver-KV key marking that `(index, repo)` serves `digest`. +/// The driver-KV key marking that `(index, repo)` serves `digest`. The value is empty: the key's +/// presence is the authorization a by-digest read checks. fn membership_key(index: &str, repo: &str, digest: &str) -> String { format!("{MEMBERSHIP_PREFIX}{index}\u{0}{repo}\u{0}{digest}") } @@ -163,7 +179,16 @@ pub fn blob_membership_key(index: &str, repo: &str, digest: &str) -> String { /// # Errors /// Returns a store error if the write fails. pub fn put_tag(meta: &MetaStore, index: &str, repo: &str, tag: &str, digest: &str) -> Result { - meta.commit_driver_txn(|txn| Ok((txn.upsert(&tag_key(index, repo, tag), digest.as_bytes())?, Vec::new()))) + meta.commit_driver_txn(|txn| Ok((put_tag_txn(txn, index, repo, tag, digest)?, Vec::new()))) +} + +/// Point `tag` at `digest` inside an open transaction, reporting whether the searchable tag set grew, +/// so a caller can publish it atomically with a quota-reservation commit. +/// +/// # Errors +/// Returns a store error if the write fails. +pub fn put_tag_txn(txn: &mut DriverTxn, index: &str, repo: &str, tag: &str, digest: &str) -> Result { + txn.upsert(&tag_key(index, repo, tag), digest.as_bytes()) } /// Resolve a tag to its cached manifest digest. diff --git a/crates/peryx-ecosystem-oci/src/tests/quota_tests.rs b/crates/peryx-ecosystem-oci/src/tests/quota_tests.rs index de392f9e..f39a9440 100644 --- a/crates/peryx-ecosystem-oci/src/tests/quota_tests.rs +++ b/crates/peryx-ecosystem-oci/src/tests/quota_tests.rs @@ -1,5 +1,16 @@ +//! Hosted-push quota enforcement: byte, project, and version reservations bracket the blob-upload, +//! cross-repo-mount, and manifest-publication boundaries, with audit mode recording rather than +//! denying. + +use std::sync::Arc; + +use axum::http::{Method, StatusCode}; +use peryx_identity::IndexAcl; +use peryx_index::{Index, IndexKind}; +use peryx_policy::{Policy, PolicyConfig}; use peryx_storage::meta::{AccountingClass, NewQuotaReservation}; +use super::{app_with, auth, body_has_code, oci_digest, send, send_body}; use crate::quota_reservation; #[test] @@ -33,3 +44,488 @@ fn test_quota_reservation_preserves_oci_identity() { ); } } + +const TOKEN: &str = "s3cret"; +const MANIFEST_TYPE: &str = "application/vnd.oci.image.manifest.v1+json"; + +/// A hosted store at route `store` whose one upload token is `s3cret`, under the quota `limits`. +fn quota_store(dir: &tempfile::TempDir, limits: &PolicyConfig) -> (Arc, axum::Router) { + let index = Index { + acl: IndexAcl::upload_token(TOKEN), + policy: Policy::compile(limits, str::to_owned), + ..super::oci_index("store", "store", IndexKind::Hosted { volatile: true }) + }; + app_with(dir, index) +} + +/// Push a blob monolithically under the upload token, returning the response status. +async fn push_blob(app: &axum::Router, repo: &str, blob: &[u8]) -> StatusCode { + let digest = oci_digest(blob); + send_body( + app, + Method::POST, + &format!("/v2/{repo}/blobs/uploads/?digest={digest}"), + &[("authorization", &auth(TOKEN))], + blob.to_vec(), + ) + .await + .0 +} + +/// Push a manifest by tag under the upload token, returning the response status. +async fn push_manifest(app: &axum::Router, repo: &str, tag: &str, body: &[u8]) -> StatusCode { + send_body( + app, + Method::PUT, + &format!("/v2/{repo}/manifests/{tag}"), + &[("authorization", &auth(TOKEN)), ("content-type", MANIFEST_TYPE)], + body.to_vec(), + ) + .await + .0 +} + +#[tokio::test] +async fn test_blob_push_over_the_repository_byte_quota_is_denied() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(4), + ..PolicyConfig::default() + }, + ); + let blob = b"five!"; + let digest = oci_digest(blob); + let (status, _, body) = send_body( + &app, + Method::POST, + &format!("/v2/store/app/blobs/uploads/?digest={digest}"), + &[("authorization", &auth(TOKEN))], + blob.to_vec(), + ) + .await; + + // The push is refused through the distribution-spec error contract, the bytes never become a + // member of the repository, and no capacity is charged. + assert_eq!(status, StatusCode::FORBIDDEN); + assert!(body_has_code(&body, "DENIED"), "{body:?}"); + assert_eq!( + send(&app, Method::GET, &format!("/v2/store/app/blobs/{digest}")) + .await + .0, + StatusCode::NOT_FOUND + ); + assert_eq!(state.meta.quota_usage("store").unwrap().accounted_bytes.committed, 0); +} + +#[tokio::test] +async fn test_blob_push_within_the_repository_byte_quota_is_accounted() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(64), + ..PolicyConfig::default() + }, + ); + let blob = b"a-real-layer-of-bytes"; + + assert_eq!(push_blob(&app, "store/app", blob).await, StatusCode::CREATED); + let usage = state.meta.quota_usage("store").unwrap(); + assert_eq!( + (usage.accounted_bytes.committed, usage.projects.committed), + (blob.len() as u64, 1) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_concurrent_push_of_one_digest_charges_bytes_once() { + let dir = tempfile::tempdir().unwrap(); + let blob = b"a-shared-layer"; + let (state, app) = quota_store( + &dir, + &PolicyConfig { + // Exactly one copy of the blob fits; deduplication must let both racing pushes in. + max_accounted_bytes: Some(blob.len() as u64), + ..PolicyConfig::default() + }, + ); + let one = tokio::spawn({ + let app = app.clone(); + async move { push_blob(&app, "store/app", blob).await } + }); + let two = tokio::spawn(async move { push_blob(&app, "store/app", blob).await }); + let (one, two) = (one.await.unwrap(), two.await.unwrap()); + + assert_eq!((one, two), (StatusCode::CREATED, StatusCode::CREATED)); + assert_eq!( + state.meta.quota_usage("store").unwrap().accounted_bytes.committed, + blob.len() as u64 + ); +} + +#[tokio::test] +async fn test_repeated_push_of_one_digest_charges_bytes_once() { + let dir = tempfile::tempdir().unwrap(); + let blob = b"a-shared-layer"; + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(blob.len() as u64), + ..PolicyConfig::default() + }, + ); + + assert_eq!(push_blob(&app, "store/app", blob).await, StatusCode::CREATED); + // A digest the repository already serves is not reserved a second time, so a re-push neither + // fails against the exact-fit limit nor inflates the logical byte counter. + assert_eq!(push_blob(&app, "store/app", blob).await, StatusCode::CREATED); + let usage = state.meta.quota_usage("store").unwrap(); + assert_eq!( + (usage.accounted_bytes.committed, usage.file_bytes.committed), + (blob.len() as u64, blob.len() as u64) + ); +} + +#[tokio::test] +async fn test_failed_blob_upload_releases_its_reservation() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(64), + ..PolicyConfig::default() + }, + ); + let wrong = format!("sha256:{}", "0".repeat(64)); + let (status, _, body) = send_body( + &app, + Method::POST, + &format!("/v2/store/app/blobs/uploads/?digest={wrong}"), + &[("authorization", &auth(TOKEN))], + b"mismatched".to_vec(), + ) + .await; + + // The commit fails on the digest mismatch, so the reservation it took is released and the + // repository is charged nothing. + assert_eq!(status, StatusCode::BAD_REQUEST); + assert!(body_has_code(&body, "DIGEST_INVALID"), "{body:?}"); + let usage = state.meta.quota_usage("store").unwrap(); + assert_eq!( + ( + usage.accounted_bytes.committed, + usage.accounted_bytes.reserved, + usage.file_bytes.committed, + usage.file_bytes.reserved, + ), + (0, 0, 0, 0) + ); +} + +#[tokio::test] +async fn test_audit_mode_records_the_violation_and_accepts_the_push() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(4), + quota_audit: true, + ..PolicyConfig::default() + }, + ); + let blob = b"a-real-layer-of-bytes"; + let digest = oci_digest(blob); + + // The same content that enforce mode denies is admitted, served, and counted, so an operator can + // observe projected enforcement against live traffic. + assert_eq!(push_blob(&app, "store/app", blob).await, StatusCode::CREATED); + assert_eq!( + send(&app, Method::GET, &format!("/v2/store/app/blobs/{digest}")) + .await + .0, + StatusCode::OK + ); + assert_eq!( + state.meta.quota_usage("store").unwrap().accounted_bytes.committed, + blob.len() as u64 + ); +} + +#[tokio::test] +async fn test_manifest_over_the_version_quota_stays_absent_from_discovery() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_versions_per_project: Some(1), + ..PolicyConfig::default() + }, + ); + let first = br#"{"schemaVersion":2}"#; + let second = br#"{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}"#; + assert_eq!(push_manifest(&app, "store/app", "v1", first).await, StatusCode::CREATED); + assert_eq!( + push_manifest(&app, "store/app", "v2", second).await, + StatusCode::FORBIDDEN + ); + + // The rejected tag reaches neither the tag listing nor by-digest discovery. + let (_, _, tags) = send(&app, Method::GET, "/v2/store/app/tags/list").await; + let tags = std::str::from_utf8(&tags).unwrap(); + assert!(tags.contains("\"v1\"") && !tags.contains("\"v2\""), "{tags:?}"); + assert_eq!( + send(&app, Method::GET, "/v2/store/app/manifests/v2").await.0, + StatusCode::NOT_FOUND + ); + assert_eq!( + send( + &app, + Method::GET, + &format!("/v2/store/app/manifests/{}", oci_digest(second)) + ) + .await + .0, + StatusCode::NOT_FOUND + ); + assert_eq!( + state + .meta + .quota_project_usage("store", "app") + .unwrap() + .versions + .committed, + 1 + ); +} + +#[tokio::test] +async fn test_manifest_re_push_under_the_same_tag_is_not_double_counted() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_versions_per_project: Some(1), + ..PolicyConfig::default() + }, + ); + let manifest = br#"{"schemaVersion":2}"#; + + // Pushing the identical image under the same tag twice is idempotent: the second push accounts no + // fresh version and so does not exhaust the single-version quota. + assert_eq!( + push_manifest(&app, "store/app", "v1", manifest).await, + StatusCode::CREATED + ); + assert_eq!( + push_manifest(&app, "store/app", "v1", manifest).await, + StatusCode::CREATED + ); + assert_eq!( + state + .meta + .quota_project_usage("store", "app") + .unwrap() + .versions + .committed, + 1 + ); +} + +#[tokio::test] +async fn test_a_new_tag_on_an_existing_manifest_counts_a_version() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_versions_per_project: Some(2), + ..PolicyConfig::default() + }, + ); + let manifest = br#"{"schemaVersion":2}"#; + + // The same digest under a second tag is a new version even though its bytes are already stored. + assert_eq!( + push_manifest(&app, "store/app", "v1", manifest).await, + StatusCode::CREATED + ); + assert_eq!( + push_manifest(&app, "store/app", "v2", manifest).await, + StatusCode::CREATED + ); + assert_eq!( + state + .meta + .quota_project_usage("store", "app") + .unwrap() + .versions + .committed, + 2 + ); +} + +#[tokio::test] +async fn test_manifest_re_push_by_digest_is_not_re_accounted() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(64), + ..PolicyConfig::default() + }, + ); + let manifest = br#"{"schemaVersion":2}"#; + let digest = oci_digest(manifest); + + assert_eq!( + push_manifest(&app, "store/app", &digest, manifest).await, + StatusCode::CREATED + ); + assert_eq!( + push_manifest(&app, "store/app", &digest, manifest).await, + StatusCode::CREATED + ); + assert_eq!( + state.meta.quota_usage("store").unwrap().accounted_bytes.committed, + manifest.len() as u64 + ); +} + +/// `POST ?mount=&from=` — publish a stored blob into `store/target` without a +/// transfer, returning the response status. +async fn mount(app: &axum::Router, digest: &str, source: &str) -> StatusCode { + send_body( + app, + Method::POST, + &format!("/v2/store/target/blobs/uploads/?mount={digest}&from={source}"), + &[("authorization", &auth(TOKEN))], + Vec::new(), + ) + .await + .0 +} + +#[tokio::test] +async fn test_cross_repo_mount_is_accounted_as_a_new_project() { + let dir = tempfile::tempdir().unwrap(); + let blob = b"a-real-layer-of-bytes"; + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_projects: Some(2), + ..PolicyConfig::default() + }, + ); + let digest = oci_digest(blob); + assert_eq!(push_blob(&app, "store/source", blob).await, StatusCode::CREATED); + + // The mount publishes the blob into a second repository and counts that repository as a project, + // even though the deduplicated bytes were already accounted. + assert_eq!(mount(&app, &digest, "store/source").await, StatusCode::CREATED); + assert_eq!( + send(&app, Method::GET, &format!("/v2/store/target/blobs/{digest}")) + .await + .0, + StatusCode::OK + ); + assert_eq!(state.meta.quota_usage("store").unwrap().projects.committed, 2); +} + +#[tokio::test] +async fn test_re_mount_of_a_present_blob_is_not_re_accounted() { + let dir = tempfile::tempdir().unwrap(); + let blob = b"a-real-layer-of-bytes"; + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_projects: Some(2), + ..PolicyConfig::default() + }, + ); + let digest = oci_digest(blob); + assert_eq!(push_blob(&app, "store/source", blob).await, StatusCode::CREATED); + assert_eq!(mount(&app, &digest, "store/source").await, StatusCode::CREATED); + + // A digest the target repository already serves is not reserved again, so re-mounting it neither + // fails against the exhausted project quota nor counts a further project. + assert_eq!(mount(&app, &digest, "store/source").await, StatusCode::CREATED); + assert_eq!(state.meta.quota_usage("store").unwrap().projects.committed, 2); +} + +#[tokio::test] +async fn test_cross_repo_mount_over_the_project_quota_is_denied() { + let dir = tempfile::tempdir().unwrap(); + let blob = b"a-real-layer-of-bytes"; + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_projects: Some(1), + ..PolicyConfig::default() + }, + ); + let digest = oci_digest(blob); + assert_eq!(push_blob(&app, "store/source", blob).await, StatusCode::CREATED); + + // The one project slot is spent on `source`, so mounting into a second repository is refused and + // that repository never comes to serve the blob. + assert_eq!(mount(&app, &digest, "store/source").await, StatusCode::FORBIDDEN); + assert_eq!( + send(&app, Method::GET, &format!("/v2/store/target/blobs/{digest}")) + .await + .0, + StatusCode::NOT_FOUND + ); + assert_eq!(state.meta.quota_usage("store").unwrap().projects.committed, 1); +} + +#[tokio::test] +async fn test_quota_decisions_increment_the_admitted_and_rejected_counters() { + let dir = tempfile::tempdir().unwrap(); + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_accounted_bytes: Some(4), + ..PolicyConfig::default() + }, + ); + assert_eq!(push_blob(&app, "store/app", b"ok").await, StatusCode::CREATED); + assert_eq!(push_blob(&app, "store/app", b"too-large").await, StatusCode::FORBIDDEN); + + let want = std::collections::BTreeMap::from([("quota_admitted", 1), ("quota_rejected", 1)]); + for _ in 0..500 { + let counters = state.metrics.index_totals(); + if counters + .get("store") + .is_some_and(|store| want.iter().all(|(key, value)| store.ecosystem.get(key) == Some(value))) + { + return; + } + std::thread::sleep(std::time::Duration::from_millis(2)); + } + panic!( + "quota metrics never settled: {:?}", + state.metrics.index_totals().get("store") + ); +} + +#[tokio::test] +async fn test_a_push_to_an_unmetered_index_records_no_quota_usage() { + let dir = tempfile::tempdir().unwrap(); + // Only a per-file size limit is set, which the byte stream enforces on its own, so no repository + // accounting runs. + let (state, app) = quota_store( + &dir, + &PolicyConfig { + max_file_size_bytes: Some(1024), + ..PolicyConfig::default() + }, + ); + + assert_eq!( + push_blob(&app, "store/app", b"a-real-layer-of-bytes").await, + StatusCode::CREATED + ); + let usage = state.meta.quota_usage("store").unwrap(); + assert_eq!((usage.accounted_bytes.committed, usage.projects.committed), (0, 0)); +} diff --git a/crates/peryx-policy/src/lib.rs b/crates/peryx-policy/src/lib.rs index 97ceb579..6894e9c9 100644 --- a/crates/peryx-policy/src/lib.rs +++ b/crates/peryx-policy/src/lib.rs @@ -22,6 +22,14 @@ pub struct PolicyConfig { pub protected_names: Vec, pub max_file_size_bytes: Option, pub max_project_size_bytes: Option, + /// Deduplicated bytes a repository may hold, enforced through the quota accounting substrate. + pub max_accounted_bytes: Option, + /// Distinct project identities a repository may hold. + pub max_projects: Option, + /// Versions a single project may hold. + pub max_versions_per_project: Option, + /// Record a would-reject quota decision instead of denying the write. + pub quota_audit: bool, } impl PolicyConfig { @@ -33,6 +41,10 @@ impl PolicyConfig { "protected_names", "max_file_size_bytes", "max_project_size_bytes", + "max_accounted_bytes", + "max_projects", + "max_versions_per_project", + "quota_audit", ]; } @@ -219,6 +231,10 @@ pub struct Policy { protected_names: ProtectedNames, max_file_size_bytes: Option, max_project_size_bytes: Option, + max_accounted_bytes: Option, + max_projects: Option, + max_versions_per_project: Option, + quota_audit: bool, rules: Vec>, recorder: Option>, active: bool, @@ -241,6 +257,10 @@ impl Policy { protected_names: ProtectedNames::compile(&config.protected_names, &normalize), max_file_size_bytes: config.max_file_size_bytes, max_project_size_bytes: config.max_project_size_bytes, + max_accounted_bytes: config.max_accounted_bytes, + max_projects: config.max_projects, + max_versions_per_project: config.max_versions_per_project, + quota_audit: config.quota_audit, rules: Vec::new(), recorder: None, active: false, @@ -283,6 +303,41 @@ impl Policy { self.max_project_size_bytes } + /// The deduplicated repository byte quota, if any. + #[must_use] + pub const fn max_accounted_bytes(&self) -> Option { + self.max_accounted_bytes + } + + /// The distinct-project quota, if any. + #[must_use] + pub const fn max_projects(&self) -> Option { + self.max_projects + } + + /// The per-project version quota, if any. + #[must_use] + pub const fn max_versions_per_project(&self) -> Option { + self.max_versions_per_project + } + + /// Whether quota decisions record a violation instead of denying the write. + #[must_use] + pub const fn quota_audit(&self) -> bool { + self.quota_audit + } + + /// Whether a write path must account against the repository quota: a repository, project, or + /// version limit is set, or audit mode observes projected enforcement. The per-file size limit is + /// enforced on the byte stream itself, so it alone does not turn accounting on. + #[must_use] + pub const fn enforces_quota(&self) -> bool { + self.max_accounted_bytes.is_some() + || self.max_projects.is_some() + || self.max_versions_per_project.is_some() + || self.quota_audit + } + /// The source policy contributed by this ecosystem, or the compatibility-preserving fallback /// mode when it contributes none. #[must_use] @@ -299,6 +354,7 @@ impl Policy { || !self.protected_names.is_empty() || self.max_file_size_bytes.is_some() || self.max_project_size_bytes.is_some() + || self.enforces_quota() || !self.rules.is_empty() } diff --git a/crates/peryx-policy/tests/policy.rs b/crates/peryx-policy/tests/policy.rs index 5c333b88..56e4016f 100644 --- a/crates/peryx-policy/tests/policy.rs +++ b/crates/peryx-policy/tests/policy.rs @@ -80,6 +80,48 @@ fn protection_matches_after_normalization() { assert!(policy.check_project(PolicyAction::Cached, "team-alpha").is_err()); } +#[test] +fn quota_limits_read_back_and_activate_the_policy() { + let policy = Policy::compile( + &PolicyConfig { + max_accounted_bytes: Some(1024), + max_projects: Some(8), + max_versions_per_project: Some(16), + quota_audit: true, + ..PolicyConfig::default() + }, + str::to_owned, + ); + + assert_eq!( + ( + policy.max_accounted_bytes(), + policy.max_projects(), + policy.max_versions_per_project(), + policy.quota_audit(), + policy.enforces_quota(), + policy.active(), + ), + (Some(1024), Some(8), Some(16), true, true, true) + ); +} + +#[test] +fn a_per_file_size_limit_alone_does_not_enforce_a_quota() { + let policy = Policy::compile( + &PolicyConfig { + max_file_size_bytes: Some(64), + ..PolicyConfig::default() + }, + str::to_owned, + ); + + // The byte stream enforces the per-file limit directly, so it does not switch repository + // accounting on, yet the policy is still active for that limit. + assert!(!policy.enforces_quota()); + assert!(policy.active()); +} + #[test] fn a_fallback_mode_renders_its_configured_wire_name() { for (mode, name) in [ diff --git a/crates/peryx/src/operator/snapshot.rs b/crates/peryx/src/operator/snapshot.rs index 90a619f0..477073ae 100644 --- a/crates/peryx/src/operator/snapshot.rs +++ b/crates/peryx/src/operator/snapshot.rs @@ -603,6 +603,10 @@ fn snapshot_policy(config: &PolicyConfig, ecosystem: &Table) -> anyhow::Result anyhow::Result