crates/gunnar-server/src/control_plane/clan_plane.rs
//! **Clans on the real plane** — [`gunnar_grpc::clans::ClanPlane`] for
//! [`CoreControlPlane`] (owner, 2026-10-05). The clan rows are
//! `gunnar_core::AuthTable`'s, in one compare-and-swap with the grants rule 2½
//! widens; the clan's namespace is the catalog's, claimed FIRST — one claim for
//! the name, exactly as a principal's account namespace is claimed with it —
//! and given back if the table refuses the clan.
//!
//! **Round 2** (owner, 2026-10-05): a member whose paid period has ended
//! LAPSES — its share stays on the row and the door stops counting it
//! ([`CoreControlPlane::clan_lapsed`], the entitlement slot's one reading);
//! and a repository moves OUT of a clan into an account, offered by the
//! clan's admin and taken by the account's admin, through the write paths
//! every push takes ([`CoreControlPlane::move_repository`]) so the twin and
//! a restore see it as they see everything else.
use std::sync::Arc;
use anyhow::Context as _;
use gunnar_core::{AccountRoom, ClanRefusal, ClanRole};
use gunnar_grpc::clans::{ClanPlane, ClanRoom};
use gunnar_grpc::seam::{ControlError, ControlResult, DeleteOptions};
use gunnar_grpc::ControlPlane as _;
use gunnar_lfs::LfsStore as _;
use gunnar_store::git_storage_trait::{Caps, RefUpdate};
use super::{internal, CoreControlPlane};
/// The table's clan onto the wire: what the plane knows. The service fills
/// what the caller sees of it. `lapsed`: the members whose shares are out of
/// force now.
fn clan_out(clan: &gunnar_core::Clan, lapsed: &[String]) -> gunnar_proto::Clan {
gunnar_proto::Clan {
name: clan.name.clone(),
created_by: clan.created_by.clone(),
created_at_unix_ms: clan.created_at,
members: clan
.members
.iter()
.map(|m| gunnar_proto::ClanMember {
principal_id: m.principal.clone(),
admin: m.admin,
contribution_bytes: m.contribution_bytes,
joined_at_unix_ms: m.joined_at,
lapsed: lapsed.contains(&m.principal),
})
.collect(),
invites: clan
.invites
.iter()
.map(|i| gunnar_proto::ClanInvite {
principal_id: i.principal.clone(),
invited_by: i.by.clone(),
at_unix_ms: i.at,
})
.collect(),
room_bytes: clan.room_bytes(),
room_in_force_bytes: clan.room_in_force(lapsed),
offers: clan
.offers
.iter()
.map(|o| offer_out(&clan.name, o, None))
.collect(),
..Default::default()
}
}
/// An offer onto the wire.
fn offer_out(
clan: &str,
offer: &gunnar_core::ClanRepoOffer,
stored: Option<u64>,
) -> gunnar_proto::RepoOffer {
gunnar_proto::RepoOffer {
clan: clan.to_owned(),
repo: offer.repo.clone(),
to_account: offer.to.clone(),
offered_by: offer.by.clone(),
at_unix_ms: offer.at,
stored_bytes: stored,
}
}
fn room_in(room: ClanRoom) -> AccountRoom {
AccountRoom {
cap: room.cap,
stored: room.stored,
}
}
/// A refusal the table made, as the seam says it.
fn clan_error(err: gunnar_core::Error) -> ControlError {
match err {
gunnar_core::Error::Clan {
clan,
refusal,
reason,
} => match refusal {
ClanRefusal::NotFound => {
let _ = reason;
ControlError::NotFound(format!("clan {clan} (none you are in or invited to)"))
}
ClanRefusal::Denied => ControlError::PermissionDenied(reason),
ClanRefusal::Precondition => ControlError::FailedPrecondition(reason),
ClanRefusal::Invalid => ControlError::InvalidArgument(reason),
ClanRefusal::Exists => ControlError::AlreadyExists(reason),
},
gunnar_core::Error::PrincipalExists(id) => {
ControlError::AlreadyExists(format!("an account called {id}"))
}
other => internal(other),
}
}
impl CoreControlPlane {
/// After a clan write: the identity half of the metadata mirror, as
/// every other identity write does.
fn clan_written(&self, clan: gunnar_core::Clan) -> gunnar_proto::Clan {
self.stores.mirror_identity();
self.clan_view(&clan)
}
/// The clan onto the wire, its lapsed shares and its offers' sizes read.
fn clan_view(&self, clan: &gunnar_core::Clan) -> gunnar_proto::Clan {
let mut out = clan_out(clan, &self.clan_lapsed(clan));
if !out.offers.is_empty() {
for offer in &mut out.offers {
offer.stored_bytes = self.repo_stored(&clan.name, &offer.repo);
}
}
out
}
/// **The members whose shares are out of force now** (round 2, owner
/// 2026-10-05: "the clan KEEPS that member's room share until the
/// member's paid period ends, then the share drops"). The entitlement
/// slot reads each member's fact — the one reading the door makes too
/// ([`crate::entitlement::EntitlementSlot::lapsed_members`]). A box that
/// sells nothing holds no paid period, and no share lapses there.
pub(crate) fn clan_lapsed(&self, clan: &gunnar_core::Clan) -> Vec<String> {
#[cfg(feature = "metered")]
if let Some(slot) = self.entitlement.as_ref() {
return slot.lapsed_members(clan);
}
let _ = clan;
Vec::new()
}
/// [`Self::clan_lapsed`] by the clan's name (empty for no clan).
fn lapsed_in(&self, clan: &str) -> ControlResult<Vec<String>> {
Ok(self
.stores
.auth()
.clan(clan)
.map_err(internal)?
.map(|row| self.clan_lapsed(&row))
.unwrap_or_default())
}
/// What `namespace/repo` stores — packs and LFS objects — or `None`
/// when it could not be measured.
fn repo_stored(&self, namespace: &str, repo: &str) -> Option<u64> {
let rows = self.repo_usage(namespace).ok()?;
rows.into_iter()
.find(|row| row.repo.as_ref().is_some_and(|p| p.repo == repo))
.map(|row| row.stored_bytes.saturating_add(row.lfs_bytes))
}
/// **Move `from` to `to`**, everything it holds, through the write paths
/// every push takes: a new repository at `to` made as `from` is (its hash,
/// default branch, visibility, description), its objects and references
/// copied in as ONE pack and ONE ref batch, HEAD, its LFS objects, its
/// branch protections and ref-name policy — then `from` purged. Every step
/// is a frame the transaction log already carries, so a twin and a restore
/// see a move as a create, a push and a purge; no new kind of frame.
///
/// A failure before the purge undoes the new repository and leaves `from`
/// exactly as it was. The pack is held in memory once.
pub(crate) fn move_repository(
&self,
from: &gunnar_proto::RepoPath,
to: &gunnar_proto::RepoPath,
by: &str,
) -> ControlResult<gunnar_proto::Repo> {
let src = self.get_repo(from)?;
if self
.stores
.catalog()
.namespace(&to.namespace)
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!("account {}", to.namespace)));
}
if self.get_repo(to).is_ok() {
return Err(ControlError::AlreadyExists(format!(
"{}/{} already exists, so {}/{} cannot move there",
to.namespace, to.repo, from.namespace, from.repo
)));
}
self.create_repo(gunnar_grpc::NewRepo {
path: to.clone(),
hash_kind: gunnar_proto::HashKind::try_from(src.hash_kind)
.unwrap_or(gunnar_proto::HashKind::Sha256),
default_branch: src.default_branch.clone(),
visibility: gunnar_proto::Visibility::try_from(src.visibility)
.unwrap_or(gunnar_proto::Visibility::Private),
description: src.description.clone(),
adopt: false,
})?;
if let Err(err) = self.copy_repository(from, to) {
if let Err(undo) = self.delete_repo(
to,
DeleteOptions {
purge: true,
retention_days: None,
principal_id: by.to_owned(),
},
) {
tracing::error!(
target: "gunnar::audit",
event = "repo.move_undo_failed",
to = %format!("{}/{}", to.namespace, to.repo),
reason = %undo,
"a failed move's new repository could not be purged again"
);
}
return Err(ControlError::Internal(format!(
"moving {}/{} to {}/{}: {err:#}; nothing was moved",
from.namespace, from.repo, to.namespace, to.repo
)));
}
// **Encrypted stays encrypted** (wiring tracer B 2026-10-08): set
// after the copy (the copy is the members' own history, older
// plaintext included, and is not re-judged) and before the purge, so a
// move never leaves a repository that takes plaintext where the
// members were promised it would not.
if src.encrypted {
if let Err(err) = self.set_encrypted(to, true) {
let _ = self.delete_repo(
to,
DeleteOptions {
purge: true,
retention_days: None,
principal_id: by.to_owned(),
},
);
return Err(ControlError::Internal(format!(
"moving {}/{} to {}/{}: its encryption could not be carried ({err}); nothing was moved",
from.namespace, from.repo, to.namespace, to.repo
)));
}
}
// The policy goes with it: a protected branch stays protected.
for rule in self.get_protection(from, None).unwrap_or_default() {
if let Err(err) = self.set_protection(to, rule) {
tracing::warn!(target: "gunnar::audit", event = "repo.move_policy_not_carried", reason = %err, "a branch protection did not move with its repository");
}
}
if let Ok(policy) = self.get_ref_name_policy(from) {
if policy != gunnar_proto::RefNamePolicy::default() {
if let Err(err) = self.set_ref_name_policy(to, policy) {
tracing::warn!(target: "gunnar::audit", event = "repo.move_policy_not_carried", reason = %err, "a ref-name policy did not move with its repository");
}
}
}
// …and the old one goes for good: its bytes leave the clan's room.
self.delete_repo(
from,
DeleteOptions {
purge: true,
retention_days: None,
principal_id: by.to_owned(),
},
)?;
self.get_repo(to)
}
/// The copy half of [`Self::move_repository`]: the snapshot's export of
/// one repository (`twin::snapshot`), written into `to` through its
/// logged store instead of into a stream.
fn copy_repository(
&self,
from: &gunnar_proto::RepoPath,
to: &gunnar_proto::RepoPath,
) -> anyhow::Result<()> {
let from_ref = gunnar_core::RepoRef::new(&from.namespace, &from.repo)?;
let to_ref = gunnar_core::RepoRef::new(&to.namespace, &to.repo)?;
let src_row = self
.stores
.catalog()
.get(&from_ref)?
.context("the repository to move is not in the catalog")?;
let dst_row = self
.stores
.catalog()
.get(&to_ref)?
.context("the new repository is not in the catalog")?;
let src = self.stores.repo_store(&src_row)?;
let dst = self.stores.repo_store(&dst_row)?;
let git = src.unlogged();
let mut updates = Vec::new();
let mut tips: Vec<Vec<u8>> = Vec::new();
for r in git.refs()? {
if r.name == "HEAD" {
continue;
}
if let Some(oid) = &r.oid {
tips.push(oid.clone());
}
updates.push(RefUpdate {
name: r.name,
target: r.oid.as_deref().map(hex::encode),
peeled: r.peeled.as_deref().map(hex::encode),
symref_target: r.symref_target,
});
}
if !tips.is_empty() {
let want: Vec<&[u8]> = tips.iter().map(Vec::as_slice).collect();
let objects = match git.select(&want, &[])? {
Some(set) => set.objects,
None => git.reachable(&want, &[])?,
};
let objs: Vec<&[u8]> = objects.iter().map(Vec::as_slice).collect();
let mut pack = Vec::new();
git.emit_pack(&objs, &[], &Caps::modern(), &mut pack)
.context("exporting the objects to move")?;
dst.serve()
.put_pack(&pack)
.context("writing the moved objects")?;
}
if !updates.is_empty() {
dst.serve()
.put_refs(&updates)
.context("writing the moved references")?;
}
if let Some(target) = git.head()?.and_then(|h| h.symref_target) {
dst.serve()
.set_head(&target)
.context("pointing the moved HEAD")?;
}
let lfs_dir = src.git_dir().join("lfs").join("objects");
let files = lfs_files(&lfs_dir)?;
if !files.is_empty() {
let store = crate::txlog_lfs::LoggedLfs::new(
gunnar_lfs::FsLfsStore::open_or_init(dst.git_dir())?,
Arc::clone(self.stores.txlog()),
format!("{}/{}", to.namespace, to.repo),
);
for path in files {
let Some(oid) = path
.file_name()
.and_then(|n| n.to_str())
.and_then(|n| gunnar_lfs::Oid::parse(n).ok())
else {
continue;
};
let mut file = std::fs::File::open(&path)
.with_context(|| format!("opening LFS {}", path.display()))?;
let size = file.metadata()?.len();
let mut pending = store.begin_write(&oid, size)?;
std::io::copy(&mut file, &mut pending)
.with_context(|| format!("copying LFS {}", path.display()))?;
pending.commit()?;
}
}
drop(src);
drop(dst);
self.stores.close_repo(&from_ref);
self.stores.close_repo(&to_ref);
Ok(())
}
}
impl ClanPlane for CoreControlPlane {
fn clan_create(
&self,
name: &str,
creator: &str,
contribution: u64,
room: ClanRoom,
) -> ControlResult<gunnar_proto::Clan> {
if name.contains('/') {
return Err(ControlError::InvalidArgument(format!(
"{name:?}: a clan's name is one namespace, with no `/` in it"
)));
}
// The namespace first — the catalog's claim on the name.
match self.stores.catalog().create_namespace(name) {
Ok(_) => {}
Err(gunnar_core::Error::NamespaceExists(_)) => {
return Err(ControlError::AlreadyExists(format!(
"a namespace called {name}"
)))
}
Err(err) => return Err(internal(err)),
}
match self
.stores
.auth()
.create_clan(name, creator, contribution, room_in(room))
{
Ok(clan) => Ok(self.clan_written(clan)),
Err(err) => {
// Made for this clan a moment ago, and empty: given back, so a
// refused clan leaves no name taken.
if let Err(undo) = self.stores.catalog().delete_namespace(name, false) {
tracing::warn!(
target: "gunnar::audit",
event = "clan.namespace_not_given_back",
clan = %name,
reason = %undo,
"a refused clan's namespace could not be removed again"
);
}
Err(clan_error(err))
}
}
}
fn clan_get(&self, name: &str) -> ControlResult<Option<gunnar_proto::Clan>> {
Ok(self
.stores
.auth()
.clan(name)
.map_err(internal)?
.as_ref()
.map(|c| self.clan_view(c)))
}
fn clans_of(&self, principal: &str) -> ControlResult<Vec<gunnar_proto::Clan>> {
Ok(self
.stores
.auth()
.clans_of(principal)
.map_err(internal)?
.iter()
.map(|c| self.clan_view(c))
.collect())
}
fn clan_invite(
&self,
clan: &str,
by: &str,
principal: &str,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.invite_to_clan(clan, by, principal)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_answer(
&self,
clan: &str,
principal: &str,
accept: bool,
contribution: u64,
room: ClanRoom,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.answer_clan_invite(clan, principal, accept, contribution, room_in(room))
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_withdraw(
&self,
clan: &str,
by: &str,
principal: &str,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.withdraw_clan_invite(clan, by, principal)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_set_contribution(
&self,
clan: &str,
member: &str,
bytes: u64,
room: ClanRoom,
used: Option<u64>,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.set_clan_contribution(
clan,
member,
bytes,
room_in(room),
used,
&self.lapsed_in(clan)?,
)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_leave(
&self,
clan: &str,
member: &str,
used: Option<u64>,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.leave_clan(clan, member, used, &self.lapsed_in(clan)?)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_remove(
&self,
clan: &str,
by: &str,
member: &str,
used: Option<u64>,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.remove_from_clan(clan, by, member, used, &self.lapsed_in(clan)?)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_set_admin(
&self,
clan: &str,
by: &str,
member: &str,
admin: bool,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.set_clan_admin(clan, by, member, admin)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
/// The admin's, and only of an EMPTY clan: no repository in its
/// namespace, live or deleted-and-still-held. The namespace goes first —
/// the catalog refuses a namespace that holds a repository, in its own
/// write, so a repository made a moment ago keeps the clan — and the clan
/// row after it.
fn clan_dissolve(&self, clan: &str, by: &str) -> ControlResult<gunnar_proto::Clan> {
let auth = self.stores.auth();
auth.clan_for(clan, by).map_err(clan_error)?;
let live = self
.stores
.catalog()
.list(Some(clan))
.map_err(internal)?
.len();
let held = gunnar_grpc::ControlPlane::list_deleted(self, Some(clan))?.len();
if live + held > 0 {
return Err(ControlError::FailedPrecondition(format!(
"clan {clan} still holds {live} repositor{} and {held} deleted one{} kept for their \
window: its admin deletes and purges them before the clan can be dissolved",
if live == 1 { "y" } else { "ies" },
if held == 1 { "" } else { "s" },
)));
}
if self
.stores
.catalog()
.namespace(clan)
.map_err(internal)?
.is_some()
{
self.stores
.catalog()
.delete_namespace(clan, false)
.map_err(internal)?;
if let Err(err) = self.stores.flow().forget(clan) {
tracing::warn!(
target: "gunnar::store",
event = "flow.forget_failed",
namespace = %clan,
%err,
"the clan's namespace was deleted but its activity series could not be removed"
);
}
}
let row = auth.dissolve_clan(clan, by).map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_room_left(&self, account: &str, room: ClanRoom) -> ControlResult<Option<u64>> {
self.stores
.auth()
.room_left(account, room_in(room))
.map_err(internal)
}
fn clan_given(&self, account: &str) -> ControlResult<u64> {
self.stores.auth().contributed_by(account).map_err(internal)
}
fn clan_offer_repo(
&self,
clan: &str,
by: &str,
repo: &str,
to: &str,
) -> ControlResult<gunnar_proto::Clan> {
// The clan's admin first — an outsider learns nothing about the
// clan's repositories — then the repository must be the clan's.
self.stores.auth().clan_for(clan, by).map_err(clan_error)?;
self.get_repo(&gunnar_proto::RepoPath::new(clan, repo))?;
let row = self
.stores
.auth()
.offer_clan_repo(clan, by, repo, to)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_repo_offer(
&self,
clan: &str,
repo: &str,
) -> ControlResult<Option<gunnar_proto::RepoOffer>> {
Ok(self
.stores
.auth()
.clan_repo_offer(clan, repo)
.map_err(internal)?
.map(|offer| offer_out(clan, &offer, self.repo_stored(clan, repo))))
}
fn clan_offers_to(&self, to: &str) -> ControlResult<Vec<gunnar_proto::RepoOffer>> {
Ok(self
.stores
.auth()
.clan_offers_to(to)
.map_err(internal)?
.iter()
.map(|(clan, offer)| offer_out(clan, offer, self.repo_stored(clan, &offer.repo)))
.collect())
}
fn clan_withdraw_offer(
&self,
clan: &str,
by: &str,
repo: &str,
) -> ControlResult<gunnar_proto::Clan> {
let row = self
.stores
.auth()
.withdraw_clan_repo_offer(clan, by, repo)
.map_err(clan_error)?;
Ok(self.clan_written(row))
}
fn clan_decline_offer(
&self,
clan: &str,
repo: &str,
to: &str,
) -> ControlResult<gunnar_proto::RepoOffer> {
let offer = self
.stores
.auth()
.close_clan_repo_offer(clan, repo, to)
.map_err(clan_error)?;
self.stores.mirror_identity();
Ok(offer_out(clan, &offer, None))
}
/// The move, then the offer closed. A move that fails leaves the offer
/// open (and the repository where it was) for another try or a decline.
fn clan_take_offer(
&self,
clan: &str,
repo: &str,
to: &str,
by: &str,
) -> ControlResult<(gunnar_proto::RepoOffer, gunnar_proto::Repo)> {
let stored = self.repo_stored(clan, repo);
let open = self
.stores
.auth()
.clan_repo_offer(clan, repo)
.map_err(internal)?
.filter(|o| o.to == to)
.ok_or_else(|| ControlError::NotFound(format!("an offer of {clan}/{repo} to {to}")))?;
let moved = self.move_repository(
&gunnar_proto::RepoPath::new(clan, repo),
&gunnar_proto::RepoPath::new(to, repo),
by,
)?;
let offer = self
.stores
.auth()
.close_clan_repo_offer(clan, repo, to)
.unwrap_or(open);
self.stores.mirror_identity();
Ok((offer_out(clan, &offer, stored), moved))
}
fn clan_role(
&self,
principal: &str,
namespace: &str,
) -> ControlResult<Option<gunnar_proto::ClanRole>> {
Ok(self
.stores
.auth()
.clan_role(principal, namespace)
.map_err(internal)?
.map(|role| match role {
ClanRole::Admin => gunnar_proto::ClanRole::Admin,
ClanRole::Member => gunnar_proto::ClanRole::Member,
ClanRole::Invited => gunnar_proto::ClanRole::Invited,
}))
}
}
/// Every LFS object file under `dir` (the snapshot's walk, `twin::snapshot`).
fn lfs_files(dir: &std::path::Path) -> anyhow::Result<Vec<std::path::PathBuf>> {
let mut out = Vec::new();
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
let entries = match std::fs::read_dir(&d) {
Ok(entries) => entries,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
Err(e) => return Err(e).with_context(|| format!("listing {}", d.display())),
};
for entry in entries {
let entry = entry?;
let kind = entry.file_type()?;
if kind.is_dir() {
stack.push(entry.path());
} else if kind.is_file() {
out.push(entry.path());
}
}
}
out.sort();
Ok(out)
}