crates/gunnar-ui-server/src/api/mirror_gate.rs
//! **Importing into your OWN namespace**: whether it fits, and how much at
//! once.
//!
//! **Who may is gunnar's** (thin layer, owner 2026-10-04): every mirror call is
//! signed AS THE PERSON ([`crate::api::Actor`]) and gunnar's own rule decides
//! (`Mirror.Add` creating a repository is `Admin` over the namespace,
//! `SetCredential`/`Remove` `Admin` on the repository, `Sync` `Write`, the
//! account gate honoured). Until then this module held a copy of gunnar's
//! grant rule (`Standing`, looser than gunnar's and blind to gates) because the
//! console signed with its own server-wide key.
//!
//! What is left is not a permission:
//!
//! 1. **Does it fit?** ([`Room`]) An import counts against the namespace's room
//! — `Usage.Read`'s `pack_bytes` against `allowance.pack_bytes`, read AS THE
//! PERSON. A namespace already at its cap imports nothing, and a repository
//! whose size the remote states (a forge listing's `size`) is refused up
//! front when it would not fit, with the customer words "needs more room"
//! and the way to buy it. The fetch itself is held to the same room by
//! gunnar (`mirror_fetch`'s pack ceiling); this check is what keeps the
//! obvious case from being queued at all.
//! 2. **How much at once?** ([`Imports`]) At most `concurrent` imports or syncs
//! in flight per person (default 2), and at most `syncs_per_hour` syncs
//! started per person (default 12). A person can only start what fits; the
//! excess is refused by name with when it can be tried again.
//!
//! # Every refusal is said once, and in plain words
//!
//! [`refused`] is the one door: it writes ONE `info` line on the audit target
//! naming the person, the namespace and the reason, and hands back a [`Fail`]
//! whose message is a sentence a customer can act on. Never `warn` — a person
//! asking for more than they may is the system working, and an operator paged
//! for it learns to ignore the page.
use std::collections::{BTreeSet, HashMap, VecDeque};
use std::sync::Mutex;
use std::time::{Duration, Instant};
use crate::api::ApiError;
use crate::app::AppState;
use crate::grpc::Fail;
use crate::pb;
/// How many imports or syncs one person may have in flight, by default.
pub const DEFAULT_CONCURRENT: usize = 2;
/// How many syncs one person may START in an hour, by default. An import's
/// first sync counts: it is the same job on the same queue.
pub const DEFAULT_SYNCS_PER_HOUR: usize = 12;
/// The window [`Limits::syncs_per_hour`] is counted over.
pub const RATE_WINDOW: Duration = Duration::from_secs(60 * 60);
/// Where a customer buys more room. The Account page's "More room" panel; on a
/// build that sells nothing there is no such panel and no link is offered.
pub const ROOM_HREF: &str = "/account";
// ───────────────────────────────────────────────────────────────────────────
// 2. Room
// ───────────────────────────────────────────────────────────────────────────
/// A namespace's room, as `Usage.Read` states it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Room {
/// `usage.pack_bytes`: what it holds now.
pub held: u64,
/// `allowance.pack_bytes`, `None` when uncapped — never spelled 0.
pub cap: Option<u64>,
/// gunnar's own `remaining.pack_bytes` (thin layer, owner 2026-10-04):
/// what the cap leaves, `None` when uncapped. Read, never computed here.
pub left: Option<u64>,
}
/// What [`Room::admits`] decided.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Fits {
/// Fits, or nothing caps it.
Yes,
/// The namespace is already at or over its cap: nothing more lands.
Full,
/// The estimate does not fit in what is left.
TooBig { left: u64 },
}
impl Room {
/// Room left under the cap, `None` when uncapped — gunnar's answer.
pub fn left(&self) -> Option<u64> {
self.left
}
/// **Would `estimate` more bytes fit?** `None` is "the remote did not say",
/// which is admitted while there is any room at all: the fetch is held to
/// the cap by gunnar as it streams.
pub fn admits(&self, estimate: Option<u64>) -> Fits {
let Some(left) = self.left() else {
return Fits::Yes;
};
if left == 0 {
return Fits::Full;
}
match estimate {
Some(bytes) if bytes > left => Fits::TooBig { left },
_ => Fits::Yes,
}
}
/// The same question after `spent` bytes of estimates have already been
/// admitted by this one import — ten repositories that each fit alone must
/// not be admitted together into room for three.
pub fn after(&self, spent: u64) -> Room {
Room {
held: self.held.saturating_add(spent),
cap: self.cap,
left: self.left.map(|left| left.saturating_sub(spent)),
}
}
}
/// Ask gunnar for `ns`'s room, with `gunnar` — the acting person's own key
/// (`Usage.Read` is a Read on the namespace) or the console's for the box's
/// administrator.
pub async fn room(gunnar: &crate::grpc::Gunnar, ns: &str) -> Result<Room, ApiError> {
let read = gunnar
.usage()
.read(pb::ReadUsageRequest {
namespace: ns.to_owned(),
})
.await
.map_err(ApiError::from)?
.into_inner();
let usage = read.usage.ok_or_else(|| {
ApiError(Fail {
code: "internal".to_owned(),
message: "the appliance answered the room question with no meters".to_owned(),
})
})?;
let cap = read
.allowance
.and_then(|a| a.pack_bytes)
// Absent means uncapped; a 0 is read the same way, the rule the push
// door applies (`caps[meter]` of 0 is "unlimited").
.filter(|cap| *cap > 0);
Ok(Room {
held: usage.pack_bytes,
cap,
// gunnar's `remaining`; only where there is a cap, as gunnar sends it.
left: cap.and(read.remaining.and_then(|r| r.pack_bytes)),
})
}
/// A byte count in words a customer reads.
pub fn bytes(count: u64) -> String {
const UNITS: &[&str] = &["B", "KiB", "MiB", "GiB", "TiB"];
let mut value = count as f64;
let mut unit = 0;
while value >= 1024.0 && unit + 1 < UNITS.len() {
value /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{count} B")
} else {
format!("{value:.1} {}", UNITS[unit])
}
}
/// Where this build sends a customer for more room, if it sells any.
pub fn room_href() -> Option<&'static str> {
#[cfg(feature = "metered")]
{
Some(ROOM_HREF)
}
#[cfg(not(feature = "metered"))]
{
None
}
}
/// **The way on after "needs more room", where nothing is sold**: the
/// community build's own line. On a build that sells, the console draws the
/// way for THIS session from the server's own words (`/auth/status`: a
/// member's `account_line`, a sold Gunnar's `shop_line`, the More room door
/// for whoever may buy) — never "add room" to somebody who does not buy here.
pub fn room_then() -> &'static str {
match room_href() {
Some(_) => "",
None => " Free some room in it, or ask whoever runs this box for more.",
}
}
/// The sentence for a refusal on room. Customer words: "needs more room" —
/// and no numbers (round 5.3, owner 2026-10-08: the customer never reads a
/// cap or the room left; they read what is needed and the way on).
pub fn room_words(ns: &str, _room: &Room, fits: Fits, what: &str, _estimate: Option<u64>) -> String {
match fits {
Fits::Full | Fits::TooBig { .. } => format!("{ns} needs more room to import {what}.{}", room_then()),
Fits::Yes => String::new(),
}
}
/// **Was this failed job stopped by the namespace's room?** gunnar's own
/// structured answer (thin layer, owner 2026-10-04): `MirrorJob.refused_reason`
/// is PACK_CAP (at the pack room, or cut at the ceiling) or PLAN (the plan
/// refuses writes at all). Until then this matched a fragment of gunnar's error
/// prose (" its pack "), which any rewording of that sentence would have broken.
pub fn stopped_at_the_cap(job: &pb::MirrorJob) -> bool {
matches!(
pb::MirrorRefusal::try_from(job.refused_reason),
Ok(pb::MirrorRefusal::PackCap | pb::MirrorRefusal::Plan)
)
}
// ───────────────────────────────────────────────────────────────────────────
// 3. Limits
// ───────────────────────────────────────────────────────────────────────────
/// The two numbers, from the environment or a test.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Limits {
/// Imports or syncs one person may have queued or running at once.
pub concurrent: usize,
/// Syncs one person may start in [`RATE_WINDOW`].
pub syncs_per_hour: usize,
}
impl Default for Limits {
fn default() -> Limits {
Limits {
concurrent: DEFAULT_CONCURRENT,
syncs_per_hour: DEFAULT_SYNCS_PER_HOUR,
}
}
}
impl Limits {
/// `GUNNAR_UI_IMPORT_CONCURRENT` and `GUNNAR_UI_SYNCS_PER_HOUR`. A value
/// that is not a positive number keeps the default and says so once at
/// start — a typo must not become "unlimited" or "never".
pub fn from_env() -> Limits {
let mut limits = Limits::default();
for (name, slot) in [
("GUNNAR_UI_IMPORT_CONCURRENT", &mut limits.concurrent),
("GUNNAR_UI_SYNCS_PER_HOUR", &mut limits.syncs_per_hour),
] {
let Ok(raw) = std::env::var(name) else {
continue;
};
match raw.trim().parse::<usize>() {
Ok(n) if n > 0 => *slot = n,
_ => tracing::warn!(
variable = name,
value = %raw,
kept = *slot,
"not a positive number; the default is kept"
),
}
}
limits
}
}
/// What a person has in flight and has started lately. In memory: a restart
/// forgets it, which errs towards letting somebody start one more.
#[derive(Debug, Default)]
struct Ledger {
/// user → paths whose job this console started and has not seen settle.
live: HashMap<String, BTreeSet<String>>,
/// user → when each recent sync was started.
started: HashMap<String, VecDeque<Instant>>,
}
/// **The per-person limits, and the bookkeeping behind them.**
#[derive(Debug, Default)]
pub struct Imports {
pub limits: Limits,
ledger: Mutex<Ledger>,
}
/// Why [`Imports::reserve`] said no.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Busy {
/// Already `running` jobs in flight, which is the most allowed.
Concurrent { running: usize, limit: usize },
/// `limit` syncs started in the window; the oldest leaves it in `wait`.
Rate { limit: usize, wait: Duration },
}
impl Busy {
/// The sentence a customer reads.
pub fn words(&self) -> String {
match self {
Busy::Concurrent { running, limit } => format!(
"You already have {running} import(s) or sync(s) running, and {limit} at a time \
is the most one account runs. This one can start when one of them finishes."
),
Busy::Rate { limit, wait } => {
let minutes = wait.as_secs().div_ceil(60).max(1);
format!(
"You have started {limit} syncs in the last hour, which is the most one \
account may start. You can start another in about {minutes} minute(s)."
)
}
}
}
}
impl Imports {
pub fn with_limits(limits: Limits) -> Imports {
Imports {
limits,
ledger: Mutex::default(),
}
}
pub fn from_env() -> Imports {
Imports::with_limits(Limits::from_env())
}
/// The paths `user` has in flight, as this console last knew them.
pub fn in_flight(&self, user: &str) -> Vec<String> {
let ledger = self.ledger.lock().expect("the import ledger");
ledger
.live
.get(user)
.map(|paths| paths.iter().cloned().collect())
.unwrap_or_default()
}
/// Forget the paths whose jobs have settled. `still_live` is what gunnar
/// says is still queued or running among [`Self::in_flight`]'s answer.
pub fn settle(&self, user: &str, still_live: &BTreeSet<String>) {
let mut ledger = self.ledger.lock().expect("the import ledger");
if let Some(paths) = ledger.live.get_mut(user) {
paths.retain(|p| still_live.contains(p));
if paths.is_empty() {
ledger.live.remove(user);
}
}
}
/// **Take one slot and one sync for `path`, or say why not.** Atomic, so
/// two tabs pressing at once cannot both take the last slot. A path that
/// is already in flight takes no new slot — gunnar dedups the job on the
/// repository, so it is the same job — but it still counts as a sync
/// started, because the rate limit is about presses.
pub fn reserve(&self, user: &str, path: &str, now: Instant) -> Result<(), Busy> {
let mut ledger = self.ledger.lock().expect("the import ledger");
let limits = self.limits;
let started = ledger.started.entry(user.to_owned()).or_default();
while started
.front()
.is_some_and(|at| now.saturating_duration_since(*at) >= RATE_WINDOW)
{
started.pop_front();
}
if started.len() >= limits.syncs_per_hour {
let oldest = *started.front().expect("non-empty: len >= limit >= 1");
let wait = RATE_WINDOW.saturating_sub(now.saturating_duration_since(oldest));
return Err(Busy::Rate {
limit: limits.syncs_per_hour,
wait,
});
}
let live = ledger.live.entry(user.to_owned()).or_default();
if !live.contains(path) && live.len() >= limits.concurrent {
return Err(Busy::Concurrent {
running: live.len(),
limit: limits.concurrent,
});
}
live.insert(path.to_owned());
ledger
.started
.entry(user.to_owned())
.or_default()
.push_back(now);
Ok(())
}
/// Give a slot back: the call it was reserved for did not enqueue a job.
/// The sync stays counted — a press that gunnar refused was still a press.
pub fn release(&self, user: &str, path: &str) {
let mut ledger = self.ledger.lock().expect("the import ledger");
if let Some(paths) = ledger.live.get_mut(user) {
paths.remove(path);
}
}
}
/// **Refresh what `user` has in flight from gunnar's own job states.** One
/// `Mirror.Status` per path this console started for them, asked as them; a path that is gone,
/// or whose job is done or failed, frees its slot. A path gunnar cannot answer
/// for keeps its slot — erring towards the limit, never past it.
pub async fn refresh(state: &AppState, gunnar: &crate::grpc::Gunnar, user: &str) {
let paths = state.imports.in_flight(user);
if paths.is_empty() {
return;
}
let mut still = BTreeSet::new();
for path in paths {
let Some((ns, name)) = path.split_once('/') else {
continue;
};
let answer = gunnar
.mirror()
.status(pb::MirrorStatusRequest {
namespace: String::new(),
repo: Some(crate::api::repo_path(ns, name)),
})
.await;
match answer {
Ok(answer) => {
let live = answer.into_inner().mirrors.into_iter().any(|m| {
m.job.is_some_and(|job| {
matches!(
pb::MirrorJobState::try_from(job.state),
Ok(pb::MirrorJobState::Queued | pb::MirrorJobState::Running)
)
})
});
if live {
still.insert(path);
}
}
Err(_) => {
still.insert(path);
}
}
}
state.imports.settle(user, &still);
}
// ───────────────────────────────────────────────────────────────────────────
// The one door every refusal goes through
// ───────────────────────────────────────────────────────────────────────────
/// The refusal codes this module mints. Not gunnar's vocabulary: these are the
/// console's own decisions, and the browser words each one (`api::Fail`).
pub mod code {
/// The namespace needs more room (403).
pub const NEEDS_ROOM: &str = "needsroom";
/// Too many at once, or too many lately (429).
pub const BUSY: &str = "busy";
}
/// **Say no, once.** One `info` line on the audit target, and the [`Fail`]
/// the browser shows. `reason` is a short machine word for the log; `words` is
/// the sentence.
pub fn refused(who: &str, namespace: &str, code: &str, reason: &str, words: String) -> ApiError {
tracing::info!(
target: crate::auth::AUDIT_TARGET,
event = "mirror.refused",
user = %who,
namespace = %namespace,
reason,
"an import or sync was refused: {words}"
);
ApiError(Fail {
code: code.to_owned(),
message: words,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn room_admits_what_fits_and_names_what_does_not() {
let uncapped = Room {
held: 5 << 30,
cap: None,
left: None,
};
assert_eq!(uncapped.admits(Some(u64::MAX)), Fits::Yes);
let room = Room {
held: 900,
cap: Some(1000),
left: Some(1000_u64.saturating_sub(900)),
};
assert_eq!(
room.admits(Some(100)),
Fits::Yes,
"exactly the room left fits"
);
assert_eq!(room.admits(Some(101)), Fits::TooBig { left: 100 });
assert_eq!(
room.admits(None),
Fits::Yes,
"an unknown size is held to the cap by gunnar"
);
let full = Room {
held: 1000,
cap: Some(1000),
left: Some(1000_u64.saturating_sub(1000)),
};
assert_eq!(full.admits(None), Fits::Full);
assert_eq!(full.admits(Some(0)), Fits::Full);
let over = Room {
held: 1200,
cap: Some(1000),
left: Some(1000_u64.saturating_sub(1200)),
};
assert_eq!(
over.admits(Some(1)),
Fits::Full,
"over the cap is full, not an underflow"
);
// Ten that each fit alone do not all fit together.
assert_eq!(room.after(60).admits(Some(60)), Fits::TooBig { left: 40 });
}
#[test]
fn the_room_refusal_says_needs_more_room() {
let room = Room {
held: 900 << 20,
cap: Some(1 << 30),
left: Some((1u64 << 30).saturating_sub(900 << 20)),
};
let words = room_words(
"kim",
&room,
room.admits(Some(512 << 20)),
"alice/linux",
Some(512 << 20),
);
assert!(words.starts_with("kim needs more room"), "{words}");
// Round 5.3: no cap, no room left, no size in the customer's words.
for never in ["MiB", "GiB", "left", "cap", "holds"] {
assert!(!words.contains(never), "{never:?}: {words}");
}
// The way on is the console's to draw for this session (a member
// asks the account's admin; a sold Gunnar's room is bought at its
// shop), never "add room" for everyone.
#[cfg(feature = "metered")]
assert!(!words.contains("Account page"), "{words}");
#[cfg(not(feature = "metered"))]
assert!(words.contains("ask whoever runs this box"), "{words}");
}
#[test]
fn a_job_stopped_at_the_cap_is_recognised_by_gunnars_own_field() {
let job = |reason: pb::MirrorRefusal, error: &str| pb::MirrorJob {
refused_reason: reason as i32,
error: error.to_owned(),
..Default::default()
};
assert!(stopped_at_the_cap(&job(pb::MirrorRefusal::PackCap, "anything at all")));
assert!(stopped_at_the_cap(&job(pb::MirrorRefusal::Plan, "")));
// The prose is not read: a sentence that names the pack but carries no
// reason is not a room stop.
assert!(!stopped_at_the_cap(&job(
pb::MirrorRefusal::Unspecified,
"plan team would pass its pack quota with this push"
)));
}
#[test]
fn at_most_n_at_once_and_the_same_repository_is_one_slot() {
let imports = Imports::with_limits(Limits {
concurrent: 2,
syncs_per_hour: 100,
});
let now = Instant::now();
assert_eq!(imports.reserve("kim", "kim/a", now), Ok(()));
assert_eq!(imports.reserve("kim", "kim/b", now), Ok(()));
assert_eq!(
imports.reserve("kim", "kim/c", now),
Err(Busy::Concurrent {
running: 2,
limit: 2
})
);
// The same repository again is the same job (gunnar dedups): no slot.
assert_eq!(imports.reserve("kim", "kim/a", now), Ok(()));
// Somebody else's slots are their own.
assert_eq!(imports.reserve("alice", "alice/x", now), Ok(()));
// One settles; the slot is free.
imports.settle("kim", &BTreeSet::from(["kim/b".to_owned()]));
assert_eq!(imports.reserve("kim", "kim/c", now), Ok(()));
// A release gives the slot back.
imports.release("kim", "kim/c");
assert_eq!(imports.in_flight("kim"), vec!["kim/b".to_owned()]);
}
#[test]
fn at_most_m_syncs_an_hour_and_the_window_slides() {
let imports = Imports::with_limits(Limits {
concurrent: 100,
syncs_per_hour: 3,
});
let t0 = Instant::now();
for i in 0..3 {
assert_eq!(
imports.reserve("kim", "kim/a", t0 + Duration::from_secs(i * 60)),
Ok(())
);
}
let at = t0 + Duration::from_secs(10 * 60);
match imports.reserve("kim", "kim/a", at) {
Err(Busy::Rate { limit: 3, wait }) => {
assert_eq!(wait, Duration::from_secs(50 * 60));
let words = Busy::Rate { limit: 3, wait }.words();
assert!(words.contains("about 50 minute"), "{words}");
}
other => panic!("expected the rate limit, got {other:?}"),
}
// An hour after the first, one sync has left the window.
assert_eq!(imports.reserve("kim", "kim/a", t0 + RATE_WINDOW), Ok(()));
}
#[test]
fn a_limit_that_is_not_a_positive_number_keeps_the_default() {
// Read through a fresh value rather than the process environment, so
// two tests in one binary cannot disagree about it.
let limits = Limits::default();
assert_eq!(limits.concurrent, 2);
assert_eq!(limits.syncs_per_hour, 12);
}
}