1use std::sync::Arc;
43
44use gunnar_core::{Grant, HashKind, Level, PrincipalKind, RepoRef, Scope, UnixMillis, Visibility};
45use gunnar_grpc::seam::ControlCounts;
46use gunnar_grpc::MergedRef;
47use gunnar_grpc::{ControlError, ControlPlane, ControlResult, NewRepo as ProtoNewRepo};
48use gunnar_proto::{
49 BranchProtection, Commit, FileChange, GrantLevel, MirrorConfig, Namespace, Principal,
50 PublicKey, PullRequest, PullRequestState, ReadLfsObjectResponse, ReadObjectResponse,
51 RefNamePolicy, RefTarget,
52 Repo, RepoPath, RepoStorage, Signature, StorageInfoResponse,
53};
54use gunnar_store::git_storage_trait::{GitServe, ObjType, RefRow};
55use gunnar_store::object;
56use gunnar_wire::RefSnapshot;
57
58use crate::storage::OpenRepo;
59use crate::stores::ServerStores;
60
61#[cfg(feature = "replication")]
76const ABDICATE_FLUSH_SECS: u64 = 30;
77
78#[cfg(feature = "replication")]
83fn abdicate_flush_secs() -> u64 {
84 resolve_flush_secs(std::env::var("GUNNAR_ABDICATE_FLUSH_SECS").ok().as_deref())
88}
89
90#[cfg(feature = "replication")]
93fn resolve_flush_secs(raw: Option<&str>) -> u64 {
94 raw.and_then(|v| v.trim().parse::<u64>().ok())
95 .filter(|&s| s > 0)
96 .unwrap_or(ABDICATE_FLUSH_SECS)
97}
98
99mod sso_plane;
101
102fn log_read_refusal(e: gunnar_core::txlog::Error) -> ControlError {
107 match e {
108 gunnar_core::txlog::Error::Released(_) => ControlError::Unavailable(format!(
109 "{} this box's data set is being grown (unmounted): its transaction log is back after the remount — ask again",
110 gunnar_grpc::PRIMARY_GROWING
111 )),
112 gunnar_core::txlog::Error::NotRetained { seq, first } => ControlError::FailedPrecondition(format!(
113 "seq {seq} is no longer retained; the oldest is {first} — a snapshot is needed, not a range"
114 )),
115 gunnar_core::txlog::Error::BeyondHead { seq, head } => ControlError::InvalidArgument(format!("seq {seq} is beyond the head {head}")),
116 other => ControlError::Internal(format!("transaction log: {other}")),
117 }
118}
119
120#[allow(clippy::type_complexity)] struct ChunkOut<'s> {
132 sink: &'s mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
133 open: Option<OutEntry>,
134}
135
136struct OutEntry {
138 seq: u64,
139 kind: u32,
140 meta: Option<Vec<u8>>,
142 blob_len: u64,
143 at: u64,
145 crc: Option<u32>,
147 ended: bool,
149}
150
151impl ChunkOut<'_> {
152 fn send(&mut self, blob: &[u8]) -> anyhow::Result<()> {
153 let Some(o) = self.open.as_mut() else {
154 anyhow::bail!("a piece outside an entry");
155 };
156 let offset = o.at;
157 o.at += blob.len() as u64;
158 anyhow::ensure!(o.at <= o.blob_len, "seq {}: more than its {} blob bytes", o.seq, o.blob_len);
159 let last = o.at == o.blob_len;
160 o.ended |= last;
161 let chunk = gunnar_proto::LogChunk {
162 seq: o.seq,
163 kind: o.kind,
164 meta: o.meta.take().unwrap_or_default(),
165 blob_len: o.blob_len,
166 blob: blob.to_vec(),
167 last,
168 restore_head: 0,
169 restore_bytes: 0,
170 log_id: String::new(),
171 offset: Some(offset),
172 entry_crc: if last { o.crc } else { None },
173 snapshot_item: None,
174 snapshot_id: String::new(),
175 };
176 (self.sink)(chunk).map_err(anyhow::Error::from)
177 }
178}
179
180impl crate::twin::FrameSink for ChunkOut<'_> {
181 fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
182 self.begin_at(seq, kind, meta, blob_len, 0)
183 }
184 fn begin_at(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64, offset: u64) -> anyhow::Result<()> {
185 anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
186 anyhow::ensure!(offset <= blob_len, "seq {seq}: from byte {offset} of a {blob_len}-byte blob");
187 self.open = Some(OutEntry { seq, kind: u32::from(kind), meta: Some(meta.to_vec()), blob_len, at: offset, crc: None, ended: false });
188 Ok(())
189 }
190 fn checked(&mut self, payload_crc: u32) -> anyhow::Result<()> {
191 if let Some(o) = self.open.as_mut() {
192 o.crc = Some(payload_crc);
193 }
194 Ok(())
195 }
196 fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
197 self.send(bytes)
198 }
199 fn end(&mut self) -> anyhow::Result<()> {
200 if matches!(&self.open, Some(o) if !o.ended) {
203 self.send(&[])?;
204 }
205 let Some(o) = self.open.take() else {
206 anyhow::bail!("an entry ended that never began");
207 };
208 anyhow::ensure!(o.at == o.blob_len && o.ended, "seq {}: {} of {} blob bytes went out", o.seq, o.at, o.blob_len);
209 Ok(())
210 }
211}
212
213impl CoreControlPlane {
214 fn readable_log(&self) -> ControlResult<&gunnar_core::txlog::Log> {
220 if crate::twin::maintenance::is_held(self.stores.root()) {
221 return Err(ControlError::Unavailable(
222 "this box's data set is held for its own growth (unmounted): its transaction log is not readable until the growth mounts it back".into(),
223 ));
224 }
225 if !self.stores.is_twin() {
226 if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
227 return Err(ControlError::FailedPrecondition(format!("{why}. Nothing is read from it")));
228 }
229 }
230 let log: &gunnar_core::txlog::Log = self.stores.txlog();
231 if log.is_released() {
232 return Err(log_read_refusal(gunnar_core::txlog::Error::Released(log.dir().to_path_buf())));
233 }
234 if self.stores.is_twin() {
237 if let Some(why) = crate::twin::snapshot::not_whole(self.stores.root()) {
238 return Err(ControlError::FailedPrecondition(format!("{why}. Nothing is read from it")));
239 }
240 }
241 Ok(log)
242 }
243}
244
245pub struct CoreControlPlane {
247 stores: Arc<ServerStores>,
248 grow: Option<Arc<crate::grow::GrowMachine>>,
252 twin_role: bool,
254 git_listener: Option<crate::readiness::GitListener>,
260 admission: Option<Arc<gunnar_ssh::Admission>>,
264 replication: crate::replication_view::ReplicationView,
277 #[cfg(feature = "replication")]
294 replication_slot: Option<Arc<crate::mirror::ReplicationSlot>>,
295 #[cfg(feature = "webhooks")]
299 webhook_slot: Option<Arc<crate::webhook::WebhookSlot>>,
300 #[cfg(feature = "metered")]
305 entitlement: Option<Arc<crate::entitlement::EntitlementSlot>>,
306 headroom: Option<Arc<gunnar_ssh::DiskHeadroomGate>>,
310 max_object_bytes: u64,
314 gc: crate::gc_queue::GcQueue,
323 purge: crate::purge_queue::PurgeQueue,
331 mirror: crate::mirror_queue::MirrorQueue,
339 bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
353 restore_keys: Vec<String>,
358}
359
360impl CoreControlPlane {
361 pub fn new(stores: Arc<ServerStores>) -> Self {
363 let for_gc = Arc::clone(&stores);
364 let bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>> = Arc::default();
365 CoreControlPlane {
366 stores,
367 grow: None,
368 twin_role: false,
369 git_listener: None,
370 admission: None,
371 replication: crate::replication_view::ReplicationView::new(),
372 #[cfg(feature = "replication")]
373 replication_slot: None,
374 #[cfg(feature = "webhooks")]
375 webhook_slot: None,
376 #[cfg(feature = "metered")]
377 entitlement: None,
378 headroom: None,
379 max_object_bytes: DEFAULT_MAX_OBJECT_BYTES,
380 gc: Self::gc_queue(Arc::clone(&for_gc)),
381 purge: Self::purge_queue(Arc::clone(&for_gc)),
382 mirror: Self::mirror_queue(Arc::clone(&for_gc), Arc::clone(&bus)),
383 bus,
384 restore_keys: Vec::new(),
385 }
386 }
387
388 pub fn with_restore_keys(mut self, keys: Vec<String>) -> Self {
392 self.restore_keys = keys;
393 self
394 }
395
396 pub fn with_bus(self, bus: Arc<gunnar_grpc::EventBus>) -> Self {
404 let _ = self.bus.set(bus);
408 self
409 }
410
411 pub fn with_twin_role(mut self, twin_role: bool) -> Self {
416 self.twin_role = twin_role;
417 self
418 }
419
420 pub fn with_git_listener(mut self, listener: crate::readiness::GitListener) -> Self {
428 self.git_listener = Some(listener);
429 self
430 }
431
432 pub fn release_stale_purge_leases(stores: &ServerStores) -> u64 {
453 let gates = match stores.auth().gates() {
454 Ok(gates) => gates,
455 Err(err) => {
456 tracing::error!(
457 target: "gunnar::audit",
458 error = %err,
459 "the account gates could not be read at start-up; any purge lease left by a \
460 crash will stay in force until it expires"
461 );
462 return 0;
463 }
464 };
465 let mut released = 0;
466 for gate in gates {
467 if gate.reason != gunnar_core::GateReason::Purge {
468 continue;
469 }
470 match stores.auth().lift_gate_held_by(&gate.account, &gate.holder) {
471 Ok(Some(_)) => {
472 released += 1;
473 tracing::warn!(
474 target: "gunnar::audit",
475 event = "account.stale_lease_released",
476 account = %gate.account,
477 job = %gate.holder,
478 "an account was still locked by a purge that did not finish; the lockout \
479 has been lifted and the account may still hold bytes the purge did not \
480 reclaim — `repo list-orphans` shows them"
481 );
482 }
483 Ok(None) => {}
484 Err(err) => tracing::error!(
485 target: "gunnar::audit",
486 account = %gate.account,
487 error = %err,
488 "a stale purge lease could not be lifted at start-up"
489 ),
490 }
491 }
492 released
493 }
494
495 fn purge_queue(stores: Arc<ServerStores>) -> crate::purge_queue::PurgeQueue {
512 let for_release = Arc::clone(&stores);
513 crate::purge_queue::PurgeQueue::new(
514 Arc::new(
515 move |order: &crate::purge_queue::PurgeOrder,
516 progress: &crate::purge_queue::Progress| {
517 let request = crate::account_purge::PurgeRequest {
518 account: order.account.clone(),
519 principal_id: order.principal_id.clone(),
520 reason: order.reason.clone(),
521 remove_principal: order.remove_principal,
522 expect_empty: order.expect_empty,
523 };
524 let auth = stores.auth();
525 let account = order.account.clone();
526 let job_id = order.job_id.clone();
527 let mut step = |step: crate::account_purge::Step| match step {
528 crate::account_purge::Step::Found(stores_found) => {
529 progress.found(stores_found)
530 }
531 crate::account_purge::Step::Reclaimed { bytes, .. } => {
532 progress.reclaimed(bytes);
533 let until = gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS;
538 if let Err(err) = auth.renew_gate(&account, &job_id, until) {
539 tracing::warn!(
540 target: "gunnar::audit",
541 account = %account,
542 job = %job_id,
543 error = %err,
544 "a purge could not renew its lockout lease; the account is \
545 still shut and the walk continues"
546 );
547 }
548 }
549 crate::account_purge::Step::Failed(note) => progress.failed(note),
550 crate::account_purge::Step::PrincipalRemoved => {
551 progress.principal_removed()
552 }
553 };
554 let report =
555 crate::account_purge::purge_account(&stores, request, &mut step)
556 .map_err(|err| format!("{err:#}"))?;
557 if report.failures.is_empty() {
562 Ok(())
563 } else {
564 Err(format!(
565 "{} of {} stores could not be reclaimed and are still on the disk; \
566 `repo list-orphans` shows them",
567 report.failures.len(),
568 report.stores_found
569 ))
570 }
571 },
572 ),
573 Arc::new(move |account: &str, job_id: &str| {
574 for_release
575 .auth()
576 .lift_gate_held_by(account, job_id)
577 .map(|lifted| lifted.is_some())
578 .map_err(|err| format!("{err}"))
579 }),
580 )
581 }
582
583 fn gc_queue(stores: Arc<ServerStores>) -> crate::gc_queue::GcQueue {
593 crate::gc_queue::GcQueue::new(Arc::new(move |path: &str| {
594 let repo_ref = gunnar_core::resolve_repo_path(path)
595 .map_err(|err| format!("`{path}` is not a valid <namespace>/<name>: {err:#}"))?;
596 let repo = stores
597 .catalog()
598 .get(&repo_ref)
599 .map_err(|err| format!("reading the catalog row for {path}: {err:#}"))?
600 .ok_or_else(|| format!("{path} is no longer a live repository"))?;
601 let report = crate::admin::gc_one(&stores, &repo)
602 .map_err(|err| format!("maintaining {path}: {err:#}"))?;
603 Ok(crate::admin::reclaimed_bytes(&report))
604 }))
605 }
606
607 fn mirror_queue(
622 stores: Arc<ServerStores>,
623 bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
624 ) -> crate::mirror_queue::MirrorQueue {
625 crate::mirror_queue::MirrorQueue::new(Arc::new(
626 move |order: &crate::mirror_queue::MirrorOrder,
627 report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>| {
628 let repo_ref = gunnar_core::resolve_repo_path(&order.path).map_err(|err| {
629 format!("`{}` is not a valid <namespace>/<name>: {err:#}", order.path)
630 })?;
631 let first = stores
638 .mirrors()
639 .mark_started(&repo_ref, gunnar_grpc::bus::now_unix_ms())
640 .ok()
641 .flatten()
642 .is_some_and(|record| record.creates_repo || record.sync_count == 0);
643 let outcome = read_credential(&stores, &repo_ref).and_then(|credential| {
644 crate::mirror_fetch::sync_once(&stores, order, &credential, report)
645 });
646 audit_mirror_outcome(&stores, &repo_ref, order, first, outcome.as_ref().err().map(String::as_str), report);
647
648 let error = outcome.as_ref().err().map(String::as_str);
654 if let Err(err) = stores.mirrors().record_sync(
655 &repo_ref,
656 gunnar_grpc::bus::now_unix_ms(),
657 error,
658 ) {
659 tracing::warn!(
660 target: "gunnar::git",
661 repo = %order.path,
662 %err,
663 "a mirror sync finished and its registration could not be updated; the \
664 fetch itself is unaffected"
665 );
666 }
667
668 if outcome.is_ok() {
672 if let Some(bus) = bus.get() {
673 let job = report.snapshot();
674 let updates = job
675 .map(|job| job.report.updates)
676 .unwrap_or_default()
677 .into_iter()
678 .map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
679 ref_name,
680 old_oid,
681 new_oid,
682 })
683 .collect::<Vec<_>>();
684 bus.publish(gunnar_proto::event::Payload::MirrorSynced(
685 gunnar_proto::MirrorSynced {
686 repo: Some(RepoPath::new(repo_ref.namespace(), repo_ref.name())),
687 remote_url: order.upstream.clone(),
688 refs_changed: updates.len() as u32,
689 updates,
690 },
691 ));
692 }
693 }
694 outcome
695 },
696 ))
697 }
698
699 pub fn resume_interrupted_mirrors(&self) -> usize {
714 let settled = match self.stores.mirrors().settle_interrupted(gunnar_grpc::bus::now_unix_ms()) {
715 Ok(settled) => settled,
716 Err(err) => {
717 tracing::warn!(target: "gunnar::git", %err, "the mirror registrations could not be read to resume interrupted syncs");
718 return 0;
719 }
720 };
721 let mut resumed = 0;
722 for record in settled {
723 let Ok(repo_ref) = gunnar_core::resolve_repo_path(&record.repo) else { continue };
724 let again = record.interrupted <= gunnar_core::MAX_RESUMES;
725 audit_line(
726 &self.stores,
727 &repo_ref,
728 if again { "mirror.sync.interrupted" } else { "mirror.sync.failed" },
729 &record.last_error,
730 true,
731 );
732 tracing::warn!(
733 target: "gunnar::git",
734 event = "mirror.sync.interrupted",
735 repo = %record.repo,
736 interrupted = record.interrupted,
737 resumed = again,
738 "a mirror sync was in flight when the server stopped"
739 );
740 if again {
741 self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
742 path: repo_ref.key(),
743 upstream: record.remote_url.clone(),
744 prune: record.prune,
745 });
746 resumed += 1;
747 }
748 }
749 resumed
750 }
751
752 fn gate_for_purge(
769 &self,
770 account: &str,
771 job_id: &str,
772 options: &gunnar_grpc::seam::PurgeAccountOptions,
773 ) -> ControlResult<()> {
774 if self
775 .stores
776 .auth()
777 .gate_covering(account)
778 .map_err(internal)?
779 .is_some()
780 {
781 return Ok(());
782 }
783 let lease = gunnar_core::AccountGate::purge_lease(
784 account,
785 job_id,
786 gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS,
787 )
788 .by(&options.principal_id)
789 .noted(format!(
790 "this account is being permanently deleted: {}",
791 options.reason
792 ));
793 self.stores.auth().gate(lease).map_err(gate_error)?;
794 Ok(())
795 }
796
797 #[must_use]
801 pub fn with_headroom(mut self, gate: Arc<gunnar_ssh::DiskHeadroomGate>) -> Self {
802 self.headroom = Some(gate);
803 self
804 }
805
806 pub fn with_grow(mut self, grow: Arc<crate::grow::GrowMachine>) -> Self {
809 self.grow = Some(grow);
810 self
811 }
812
813 pub fn with_admission(mut self, admission: Arc<gunnar_ssh::Admission>) -> Self {
816 self.admission = Some(admission);
817 self
818 }
819
820 fn grow_machine(&self) -> ControlResult<&Arc<crate::grow::GrowMachine>> {
822 self.grow.as_ref().ok_or(ControlError::Unimplemented(
823 "this control plane was built with no runtime growth wired in, so it cannot grow the data set",
824 ))
825 }
826
827 fn verify_membership_event(
837 &self,
838 signed: gunnar_proto::SignedMembershipEvent,
839 group_id: &str,
840 member_id: &str,
841 action: gunnar_core::MembershipAction,
842 ) -> ControlResult<gunnar_core::SignedMembershipEvent> {
843 let signed = from_proto_membership(signed)?;
844 if signed.event.group != group_id || signed.event.member != member_id {
845 return Err(ControlError::InvalidArgument(format!(
846 "the signed event is about {:?}/{:?} but the change is {:?}/{:?}",
847 signed.event.group, signed.event.member, group_id, member_id
848 )));
849 }
850 if signed.event.action != action {
851 return Err(ControlError::InvalidArgument(
852 "the signed event's action does not match the change (add vs remove)".to_owned(),
853 ));
854 }
855 let key = self
856 .stores
857 .auth()
858 .signing_keys()
859 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
860 .into_iter()
861 .find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
862 .ok_or_else(|| {
863 ControlError::InvalidArgument(format!(
864 "{} is not registered on this server, so nobody reading this event later \
865 could check its signature",
866 signed.key_fingerprint
867 ))
868 })?;
869 if key.principal_id != signed.event.actor {
870 return Err(ControlError::InvalidArgument(format!(
871 "this membership change is attributed to {:?} and is signed by {}, which is \
872 {:?}'s key; a membership change is the actor's own word, so it must carry the \
873 actor's own signature",
874 signed.event.actor, signed.key_fingerprint, key.principal_id
875 )));
876 }
877 let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
878 .map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
879 gunnar_core::membership::verify_event(&signed, &public)
880 .map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
881 Ok(signed)
882 }
883
884 fn append_membership_event(
889 &self,
890 signed: &gunnar_core::SignedMembershipEvent,
891 ) -> ControlResult<u64> {
892 let log = self.stores.membership_log();
893 let mut line = serde_json::to_string(signed)
894 .map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
895 line.push('\n');
896 let _framed = gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
897 .map_err(|err| ControlError::Unavailable(format!("framing the line in the transaction log: {err}")))?;
898 use std::io::Write as _;
899 let mut file = std::fs::OpenOptions::new()
900 .create(true)
901 .append(true)
902 .open(log.records_path())
903 .map_err(|err| ControlError::Unavailable(format!("opening the membership log: {err}")))?;
904 file.write_all(line.as_bytes()).map_err(|err| {
905 ControlError::Unavailable(format!("appending to the membership log: {err}"))
906 })?;
907 let count = log
908 .lines_of()
909 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
910 .len() as u64;
911 Ok(count.saturating_sub(1))
912 }
913
914 fn membership_log_events(&self) -> ControlResult<Vec<gunnar_core::SignedMembershipEvent>> {
916 self.stores
917 .membership_log()
918 .lines_of()
919 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
920 .iter()
921 .map(|line| {
922 serde_json::from_str(line).map_err(|err| {
923 ControlError::Internal(format!("decoding a membership event: {err}"))
924 })
925 })
926 .collect()
927 }
928
929 #[cfg(feature = "replication")]
937 pub fn with_replication(mut self, slot: Arc<crate::mirror::ReplicationSlot>) -> Self {
938 self.replication_slot = Some(slot);
939 self
940 }
941
942 #[cfg(feature = "replication")]
948 fn replication_slot(&self) -> ControlResult<&Arc<crate::mirror::ReplicationSlot>> {
949 self.replication_slot.as_ref().ok_or(ControlError::Unimplemented(
950 "this control plane was built with no replication slot, so it cannot configure a standby",
951 ))
952 }
953
954 #[cfg(feature = "webhooks")]
958 pub fn with_webhooks(mut self, slot: Arc<crate::webhook::WebhookSlot>) -> Self {
959 self.webhook_slot = Some(slot);
960 self
961 }
962
963 #[cfg(feature = "metered")]
965 pub fn with_entitlement(mut self, slot: Arc<crate::entitlement::EntitlementSlot>) -> Self {
966 self.entitlement = Some(slot);
967 self
968 }
969
970 #[cfg(feature = "metered")]
972 fn entitlement(&self) -> ControlResult<&Arc<crate::entitlement::EntitlementSlot>> {
973 self.entitlement.as_ref().ok_or_else(|| {
974 ControlError::FailedPrecondition(
975 "this metered server was started without `--monetize-pubkey`, so it holds no \
976 key to verify an entitlement fact against and refuses every one; every \
977 namespace is Free"
978 .into(),
979 )
980 })
981 }
982
983 #[cfg(feature = "webhooks")]
985 fn webhook_slot(&self) -> ControlResult<&Arc<crate::webhook::WebhookSlot>> {
986 self.webhook_slot.as_ref().ok_or(ControlError::Unimplemented(
987 "this control plane was built with no webhook slot, so it cannot configure an outbound endpoint",
988 ))
989 }
990
991 #[cfg(feature = "replication")]
997 fn replication_snapshot(
1005 &self,
1006 include_refs: bool,
1007 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
1008 let slot = self.replication_slot()?;
1009 let config = slot
1010 .config()
1011 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
1012 let mut out = gunnar_proto::ReplicationStatusResponse {
1013 targets: config
1014 .targets
1015 .iter()
1016 .map(crate::replication_wire::target_to_proto)
1017 .collect(),
1018 role: config.role.to_proto() as i32,
1019 ..Default::default()
1020 };
1021 if let Some(replicator) = slot.replicator() {
1022 crate::replication_wire::running_to_proto(&replicator, include_refs, &mut out);
1023 }
1024 Ok(out)
1025 }
1026
1027 pub fn replication_view(&self) -> crate::replication_view::ReplicationView {
1035 self.replication.clone()
1036 }
1037
1038 fn forge_events(&self) -> ControlResult<Vec<gunnar_core::SignedEvent>> {
1044 self.stores
1045 .forge_log()
1046 .lines_of()
1047 .map_err(|err| ControlError::Unavailable(format!("reading the forge log: {err:#}")))?
1048 .iter()
1049 .map(|line| {
1050 serde_json::from_str(line).map_err(|err| {
1051 ControlError::Internal(format!("a forge event will not decode: {err}"))
1052 })
1053 })
1054 .collect()
1055 }
1056
1057 fn review_state(
1073 &self,
1074 repo: &str,
1075 state: &gunnar_core::forge::Review,
1076 all: &[gunnar_core::SignedEvent],
1077 ) -> ControlResult<gunnar_proto::ReviewState> {
1078 let keys: Vec<(String, String, gunnar_core::ssh_key::PublicKey)> = self
1079 .stores
1080 .auth()
1081 .signing_keys()
1082 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
1083 .iter()
1084 .filter_map(|row| {
1085 gunnar_core::ssh_key::PublicKey::from_openssh(&row.openssh)
1086 .ok()
1087 .map(|key| {
1088 (
1089 row.fingerprint_sha256.clone(),
1090 row.principal_id.clone(),
1091 key,
1092 )
1093 })
1094 })
1095 .collect();
1096
1097 let approvals = state
1098 .approvals
1099 .iter()
1100 .map(|(actor, commit)| {
1101 let signed = all.iter().rev().find(|e| {
1102 e.event.repo == repo
1103 && e.event.pull == state.pull
1104 && e.event.actor == *actor
1105 && matches!(&e.event.event, gunnar_core::Event::Approved { commit: c } if c == commit)
1106 });
1107 let (verified, verdict) = match signed {
1108 None => (false, "NO EVENT".to_owned()),
1109 Some(signed) => match keys
1110 .iter()
1111 .find(|(fingerprint, _, _)| *fingerprint == signed.key_fingerprint)
1112 {
1113 None => (false, "UNKNOWN KEY".to_owned()),
1114 Some((_, owner, _)) if *owner != *actor => {
1120 (false, format!("KEY OF {owner}, NOT {actor}"))
1121 }
1122 Some((_, _, key)) => match gunnar_core::forge::verify_event(signed, key) {
1123 Ok(()) => (true, "verified".to_owned()),
1124 Err(problem) => (false, format!("BAD ({problem})")),
1125 },
1126 },
1127 };
1128 gunnar_proto::Approval {
1129 actor: actor.clone(),
1130 commit: commit.clone(),
1131 verified,
1132 verdict,
1133 }
1134 })
1135 .collect();
1136
1137 Ok(gunnar_proto::ReviewState {
1138 repo: repo.to_owned(),
1139 pull: state.pull,
1140 approvals,
1141 comments: state.comments as u64,
1142 })
1143 }
1144
1145 pub fn with_max_object_bytes(mut self, bytes: u64) -> Self {
1160 self.max_object_bytes = bytes;
1161 self
1162 }
1163
1164 fn pulls(&self) -> ControlResult<gunnar_core::Pulls> {
1169 gunnar_core::Pulls::new(Arc::clone(self.stores.control_store()))
1170 .map_err(|err| ControlError::Internal(format!("opening the pulls table: {err:#}")))
1171 }
1172
1173 fn repo_ref(path: &RepoPath) -> ControlResult<RepoRef> {
1174 path.validate()
1175 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
1176 RepoRef::new(&path.namespace, &path.repo)
1177 .map_err(|err| ControlError::InvalidArgument(err.to_string()))
1178 }
1179
1180 fn tree_of(
1191 &self,
1192 git: &dyn GitServe,
1193 hash: gix_hash::Kind,
1194 path: &RepoPath,
1195 spec: &str,
1196 ) -> ControlResult<gix_hash::ObjectId> {
1197 let id = match gix_hash::ObjectId::from_hex(spec.as_bytes()) {
1198 Ok(id) => {
1199 if id.kind() != hash {
1200 return Err(ControlError::InvalidArgument(format!(
1201 "{spec} is a {} oid; this repository is {hash}",
1202 id.kind()
1203 )));
1204 }
1205 id
1206 }
1207 Err(_) => {
1208 gunnar_store::refname::validate(spec)
1209 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
1210 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
1211 let resolved = if spec == gunnar_store::HEAD {
1212 snapshot.head(git).map_err(wire_error)?
1213 } else {
1214 snapshot.resolve(spec).map_err(wire_error)?
1215 };
1216 resolved.ok_or_else(|| ControlError::NotFound(format!("{spec} in {path}")))?
1217 }
1218 };
1219 let id = peel(git, hash, id)?;
1220 let (kind, bytes) = git
1221 .read(id.as_bytes())
1222 .map_err(store_error)?
1223 .ok_or_else(|| ControlError::NotFound(format!("object {id} in {path}")))?;
1224 match kind {
1225 ObjType::Tree => Ok(id),
1226 ObjType::Commit => object::commit(&bytes, hash)
1227 .map(|commit| commit.tree)
1228 .map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}"))),
1229 other => Err(ControlError::InvalidArgument(format!(
1232 "{spec} names a {other:?}; a diff needs a commit, a tag or a tree"
1233 ))),
1234 }
1235 }
1236
1237 fn policy_of(&self, repo: &gunnar_core::Repo) -> ControlResult<gunnar_policy::PolicyConfig> {
1247 let stored = self.stores.policies().get(&repo.path()).map_err(internal)?;
1248 match stored {
1249 None => Ok(gunnar_policy::PolicyConfig::default()),
1250 Some(document) => gunnar_policy::PolicyConfig::from_json(&document).map_err(|err| {
1251 ControlError::Internal(format!(
1252 "the stored policy for {} does not parse: {err}",
1253 repo.key()
1254 ))
1255 }),
1256 }
1257 }
1258
1259 fn store_policy(
1266 &self,
1267 repo: &gunnar_core::Repo,
1268 config: &gunnar_policy::PolicyConfig,
1269 ) -> ControlResult<()> {
1270 let policies = self.stores.policies();
1271 if *config == gunnar_policy::PolicyConfig::default() {
1272 policies.remove(&repo.path()).map_err(internal)?;
1273 return Ok(());
1274 }
1275 let json = config
1276 .to_json()
1277 .map_err(|err| ControlError::Internal(format!("rendering the policy: {err}")))?;
1278 policies.set(&repo.path(), &json).map_err(internal)
1279 }
1280
1281 fn open(&self, path: &RepoPath) -> ControlResult<(gunnar_core::Repo, Arc<OpenRepo>)> {
1282 let repo_ref = Self::repo_ref(path)?;
1283 let repo = self
1284 .stores
1285 .catalog()
1286 .get(&repo_ref)
1287 .map_err(internal)?
1288 .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
1289 let store = self.stores.repo_store(&repo).map_err(|err| {
1290 ControlError::Internal(format!("opening the repository's stores: {err:#}"))
1291 })?;
1292 Ok((repo, store))
1293 }
1294
1295 fn mirror_record(&self, path: &RepoPath) -> ControlResult<gunnar_core::MirrorRecord> {
1301 let repo_ref = Self::repo_ref(path)?;
1302 let record = self.stores.mirrors().get(&repo_ref).map_err(internal)?;
1303 match record {
1304 Some(record) if record.creates_repo && self.stores.catalog().get(&repo_ref).map_err(internal)?.is_none() => {
1305 Ok(record)
1306 }
1307 record => {
1308 self.open(path)?;
1309 record.ok_or_else(|| ControlError::NotFound(format!("no mirror registered for {path}")))
1310 }
1311 }
1312 }
1313}
1314
1315fn gc_job_out(path: &RepoPath, job: crate::gc_queue::GcJob) -> gunnar_grpc::seam::GcJob {
1323 gunnar_grpc::seam::GcJob {
1324 job_id: job.job_id,
1325 path: path.clone(),
1326 state: match job.state {
1327 crate::gc_queue::GcState::Queued => gunnar_grpc::seam::GcState::Queued,
1328 crate::gc_queue::GcState::Running => gunnar_grpc::seam::GcState::Running,
1329 crate::gc_queue::GcState::Done => gunnar_grpc::seam::GcState::Done,
1330 crate::gc_queue::GcState::Failed => gunnar_grpc::seam::GcState::Failed,
1331 },
1332 queue_position: job.queue_position,
1333 bytes_reclaimed: job.report.bytes_reclaimed,
1337 error: job.error,
1338 enqueued_at_unix_ms: job.enqueued_at_unix_ms,
1339 finished_at_unix_ms: job.finished_at_unix_ms,
1340 }
1341}
1342
1343fn gate_out(gate: gunnar_core::AccountGate) -> gunnar_grpc::seam::AccountGate {
1345 gunnar_grpc::seam::AccountGate {
1346 account: gate.account,
1347 reason: match gate.reason {
1348 gunnar_core::GateReason::Administrative => {
1349 gunnar_grpc::seam::GateReason::Administrative
1350 }
1351 gunnar_core::GateReason::Purge => gunnar_grpc::seam::GateReason::Purge,
1352 },
1353 holder: gate.holder,
1354 actor: gate.actor,
1355 note: gate.note,
1356 at_unix_ms: gate.at,
1357 expires_unix_ms: gate.expires.unwrap_or(0),
1361 }
1362}
1363
1364fn purge_job_out(job: crate::purge_queue::PurgeJob) -> gunnar_grpc::seam::PurgeJob {
1366 let report = job.report;
1370 gunnar_grpc::seam::PurgeJob {
1371 job_id: job.job_id,
1372 account: job.key,
1374 state: match job.state {
1375 crate::purge_queue::PurgeState::Queued => gunnar_grpc::seam::PurgeState::Queued,
1376 crate::purge_queue::PurgeState::Running => gunnar_grpc::seam::PurgeState::Running,
1377 crate::purge_queue::PurgeState::Done => gunnar_grpc::seam::PurgeState::Done,
1378 crate::purge_queue::PurgeState::Failed => gunnar_grpc::seam::PurgeState::Failed,
1379 },
1380 queue_position: job.queue_position,
1381 stores_found: report.stores_found,
1382 stores_purged: report.stores_purged,
1383 bytes_reclaimed: report.bytes_reclaimed,
1384 principal_removed: report.principal_removed,
1385 gate_released: report.gate_released,
1386 error: job.error,
1387 failures: report.failures,
1388 enqueued_at_unix_ms: job.enqueued_at_unix_ms,
1389 started_at_unix_ms: job.started_at_unix_ms,
1390 finished_at_unix_ms: job.finished_at_unix_ms,
1391 }
1392}
1393
1394fn gate_error(err: gunnar_core::Error) -> ControlError {
1401 match err {
1402 gunnar_core::Error::ReservedPrincipalId(_) | gunnar_core::Error::InvalidPath { .. } => {
1403 ControlError::InvalidArgument(format!("{err}"))
1404 }
1405 other => internal(other),
1406 }
1407}
1408
1409fn day_activity(day: i64, row: &gunnar_core::FlowDay) -> gunnar_proto::DayActivity {
1417 gunnar_proto::DayActivity {
1418 day_unix_ms: gunnar_core::midnight_of(day),
1419 bytes_in: row.bytes_in,
1420 bytes_out: row.bytes_out,
1421 pushes: row.pushes,
1422 fetches: row.fetches,
1423 lfs_bytes_in: row.lfs_bytes_in,
1424 lfs_bytes_out: row.lfs_bytes_out,
1425 refused: row.refused(),
1426 refused_by_reason: gunnar_core::Refusal::ALL
1427 .into_iter()
1428 .filter(|reason| row.refused_for(*reason) > 0)
1429 .map(|reason| (reason.name().to_owned(), row.refused_for(reason)))
1430 .collect(),
1431 auth_failures: row.auth_failures,
1432 last_push_unix_ms: row.last_push_unix_ms,
1433 last_fetch_unix_ms: row.last_fetch_unix_ms,
1434 distinct_principals: row.distinct_principals(),
1435 distinct_principals_capped: row.principals_capped,
1436 }
1437}
1438
1439pub(crate) fn create_error(err: &anyhow::Error, path: &RepoPath) -> ControlError {
1444 match err.downcast_ref::<gunnar_core::Error>() {
1445 Some(gunnar_core::Error::RepoExists(_)) => ControlError::AlreadyExists(format!("repository {path}")),
1446 Some(gunnar_core::Error::NoSuchNamespace(ns)) => ControlError::NotFound(format!("namespace {ns}")),
1447 Some(e @ (gunnar_core::Error::LogFollows | gunnar_core::Error::OffStripe { .. })) => ControlError::Unavailable(e.to_string()),
1448 Some(_) => ControlError::Internal(format!("{err:#}")),
1449 None => ControlError::FailedPrecondition(format!("{err:#}")),
1450 }
1451}
1452
1453#[cfg(test)]
1454mod create_error_tests {
1455 use super::*;
1456
1457 #[test]
1460 fn a_create_on_a_restoring_box_is_unavailable() {
1461 let path = RepoPath::new("acme", "site");
1462 let restoring = anyhow::Error::new(gunnar_core::Error::LogFollows);
1463 assert!(matches!(create_error(&restoring, &path), ControlError::Unavailable(m) if m.contains("being restored")));
1464 let exists = anyhow::Error::new(gunnar_core::Error::RepoExists(gunnar_core::RepoRef::new("acme", "site").unwrap()));
1465 assert!(matches!(create_error(&exists, &path), ControlError::AlreadyExists(_)));
1466 assert!(matches!(create_error(&anyhow::anyhow!("a store already stands there"), &path), ControlError::FailedPrecondition(_)));
1467 }
1468}
1469
1470fn internal(err: gunnar_core::Error) -> ControlError {
1499 use gunnar_core::Error as E;
1500 match err {
1501 e @ (E::InvalidPath { .. }
1503 | E::PathEscape { .. }
1504 | E::ReservedPrincipalId(_)
1505 | E::BadBranchName(_)
1506 | E::BadFingerprint(_)
1507 | E::Key(_)
1508 | E::InvalidInput { .. }) => ControlError::InvalidArgument(e.to_string()),
1509 e @ (E::NoSuchRepo(_)
1511 | E::NoSuchNamespace(_)
1512 | E::NoSuchPrincipal(_)
1513 | E::NoSuchCredential(_)) => ControlError::NotFound(e.to_string()),
1514 e @ (E::RepoExists(_)
1516 | E::NamespaceExists(_)
1517 | E::PrincipalExists(_)
1518 | E::NameExists { .. }
1519 | E::KeyExists { .. }
1520 | E::CredentialExists { .. }
1521 | E::InviteExists(_)
1522 | E::TokenExists { .. }
1523 | E::AlreadyBootstrapped { .. }
1524 | E::AlreadyProvisioned { .. }) => ControlError::AlreadyExists(e.to_string()),
1525 e @ (E::NamespaceNotEmpty { .. }
1529 | E::Contended { .. }
1530 | E::NotAPerson { .. }
1531 | E::NotAMachine { .. }) => {
1532 ControlError::FailedPrecondition(e.to_string())
1533 }
1534 e @ (E::Denied { .. } | E::CounterRollback { .. } | E::InviteInvalid) => {
1535 ControlError::PermissionDenied(e.to_string())
1536 }
1537 e @ (E::Store(_) | E::Json(_)) => ControlError::Unavailable(e.to_string()),
1541 e @ E::OffStripe { .. } => ControlError::Unavailable(e.to_string()),
1546 e @ E::LogFollows => ControlError::Unavailable(e.to_string()),
1548 e => ControlError::Internal(e.to_string()),
1555 }
1556}
1557
1558#[cfg(feature = "engine-znippy")]
1575fn archive_error(err: anyhow::Error) -> ControlError {
1576 let io = err
1577 .chain()
1578 .any(|cause| cause.downcast_ref::<std::io::Error>().is_some());
1579 let message = format!("{err:#}");
1580 if io {
1581 ControlError::Unavailable(message)
1582 } else {
1583 ControlError::FailedPrecondition(message)
1584 }
1585}
1586
1587fn store_error(err: anyhow::Error) -> ControlError {
1609 ControlError::Internal(format!("{err:#}"))
1610}
1611
1612fn wire_error(err: gunnar_wire::Error) -> ControlError {
1620 ControlError::Internal(format!("{err:#}"))
1621}
1622
1623pub(crate) fn hash_kind(kind: HashKind) -> gunnar_proto::HashKind {
1629 match kind {
1630 HashKind::Sha1 => gunnar_proto::HashKind::Sha1,
1631 HashKind::Sha256 => gunnar_proto::HashKind::Sha256,
1632 }
1633}
1634
1635fn visibility_in(v: gunnar_proto::Visibility) -> ControlResult<Visibility> {
1649 match v {
1650 gunnar_proto::Visibility::Public => Ok(Visibility::Public),
1651 gunnar_proto::Visibility::Internal => Err(ControlError::InvalidArgument(
1652 "visibility \"internal\" has no implementation: since 2026-08-03, \"public\" means \
1653 readable with no key at all, so \"internal\" (any authenticated principal) \
1654 no longer aliases it and gunnar refuses to guess which you meant. Choose \
1655 \"private\" or \"public\""
1656 .into(),
1657 )),
1658 gunnar_proto::Visibility::Private | gunnar_proto::Visibility::Unspecified => {
1659 Ok(Visibility::Private)
1660 }
1661 }
1662}
1663
1664fn pull_out(path: &RepoPath, record: gunnar_core::PullRecord) -> PullRequest {
1666 PullRequest {
1667 repo: Some(path.clone()),
1668 number: record.number,
1669 title: record.title,
1670 description: record.description,
1671 source_ref: record.source_ref,
1672 target_ref: record.target_ref,
1673 state: match record.state {
1674 gunnar_core::PullState::Open => PullRequestState::Open,
1675 gunnar_core::PullState::Merged => PullRequestState::Merged,
1676 gunnar_core::PullState::Closed => PullRequestState::Closed,
1677 } as i32,
1678 author: record.author,
1679 created_at_unix_ms: record.created_ms,
1680 updated_at_unix_ms: record.updated_ms,
1681 source_oid: record.source_oid,
1682 merge_oid: record.merge_oid.unwrap_or_default(),
1683 }
1684}
1685
1686fn visibility_out(v: Visibility) -> gunnar_proto::Visibility {
1687 match v {
1688 Visibility::Private => gunnar_proto::Visibility::Private,
1689 Visibility::Public => gunnar_proto::Visibility::Public,
1690 }
1691}
1692
1693fn principal_kind_in(kind: gunnar_proto::PrincipalKind) -> PrincipalKind {
1694 match kind {
1695 gunnar_proto::PrincipalKind::Machine => PrincipalKind::Machine,
1696 gunnar_proto::PrincipalKind::Deploy => PrincipalKind::Deploy,
1697 gunnar_proto::PrincipalKind::Group => PrincipalKind::Group,
1698 gunnar_proto::PrincipalKind::User | gunnar_proto::PrincipalKind::Unspecified => {
1699 PrincipalKind::User
1700 }
1701 }
1702}
1703
1704impl CoreControlPlane {
1705 fn owner_kind(&self, name: &str) -> i32 {
1716 let auth = self.stores.auth();
1717 match auth.principal(name) {
1718 Ok(Some(p)) if p.kind == PrincipalKind::Group => {
1719 let members = auth.members_of(name).unwrap_or_default();
1720 let machines = !members.is_empty() && members.iter().all(|m| matches!(auth.principal(m), Ok(Some(q)) if q.kind == PrincipalKind::Machine));
1721 if machines { gunnar_proto::PrincipalKind::Machine as i32 } else { principal_kind_out(p.kind) as i32 }
1722 }
1723 Ok(Some(p)) => principal_kind_out(p.kind) as i32,
1724 _ => gunnar_proto::PrincipalKind::Unspecified as i32,
1725 }
1726 }
1727
1728 fn namespace_out(&self, namespace: gunnar_core::Namespace, employees: u32) -> Namespace {
1733 Namespace {
1734 owner_kind: self.owner_kind(&namespace.name),
1735 owner_server_grant: self.stores.auth().holds_server_grant(&namespace.name).unwrap_or(true),
1738 name: namespace.name,
1739 description: String::new(),
1740 created_at_unix_ms: 0,
1741 public: namespace.public,
1742 events_public: namespace.events_public,
1743 employees,
1744 }
1745 }
1746
1747 fn employee_count(&self, account: &str) -> ControlResult<u32> {
1749 Ok(self
1750 .stores
1751 .auth()
1752 .employees_of(account)
1753 .map_err(internal)?
1754 .len() as u32)
1755 }
1756}
1757
1758fn principal_kind_out(kind: PrincipalKind) -> gunnar_proto::PrincipalKind {
1759 match kind {
1760 PrincipalKind::User => gunnar_proto::PrincipalKind::User,
1761 PrincipalKind::Machine => gunnar_proto::PrincipalKind::Machine,
1762 PrincipalKind::Deploy => gunnar_proto::PrincipalKind::Deploy,
1763 PrincipalKind::Group => gunnar_proto::PrincipalKind::Group,
1764 PrincipalKind::Anonymous => gunnar_proto::PrincipalKind::Unspecified,
1773 }
1774}
1775
1776fn level_in(level: GrantLevel) -> ControlResult<Level> {
1777 match level {
1778 GrantLevel::Read => Ok(Level::Read),
1779 GrantLevel::Write => Ok(Level::Write),
1780 GrantLevel::Admin => Ok(Level::Admin),
1781 GrantLevel::Unspecified => Err(ControlError::InvalidArgument(
1782 "grant level unspecified".into(),
1783 )),
1784 }
1785}
1786
1787fn level_out(level: Level) -> GrantLevel {
1788 match level {
1789 Level::Read => GrantLevel::Read,
1790 Level::Write => GrantLevel::Write,
1791 Level::Admin => GrantLevel::Admin,
1792 }
1793}
1794
1795fn repo_out(repo: &gunnar_core::Repo) -> Repo {
1796 Repo {
1797 path: Some(RepoPath::new(&repo.namespace, &repo.name)),
1798 hash_kind: hash_kind(repo.hash_kind) as i32,
1799 default_branch: repo.default_branch.clone(),
1800 visibility: visibility_out(repo.visibility) as i32,
1801 created_at_unix_ms: repo.created,
1802 description: repo.description.clone(),
1803 encrypted: repo.encrypted,
1804 }
1805}
1806
1807fn web_credential_out(row: &gunnar_core::WebCredential) -> ControlResult<gunnar_proto::WebCredential> {
1809 let credential_id = gunnar_grpc::webauthn::from_b64url(&row.id).ok_or_else(|| {
1810 ControlError::Internal(format!("a stored web credential has an id that is not base64url: {}", row.id))
1811 })?;
1812 Ok(gunnar_proto::WebCredential {
1813 credential_id,
1814 principal_id: row.principal_id.clone(),
1815 label: row.label.clone(),
1816 created_at_unix_ms: row.created,
1817 last_used_unix_ms: row.last_used,
1818 sign_count: row.sign_count,
1819 })
1820}
1821
1822fn level_in_opt(level: GrantLevel) -> Option<Level> {
1824 level_in(level).ok()
1825}
1826
1827fn invite_out(row: &gunnar_core::Invite) -> gunnar_proto::Invite {
1829 gunnar_proto::Invite {
1830 invite_id: row.id.clone(),
1831 principal_id: row.principal_id.clone(),
1832 created_by: row.created_by.clone(),
1833 created_at_unix_ms: row.created,
1834 expires_at_unix_ms: row.expires,
1835 note: row.note.clone(),
1836 server_level: row.server_level.map(level_out).unwrap_or(GrantLevel::Unspecified) as i32,
1837 }
1838}
1839
1840fn principal_out(principal: &gunnar_core::Principal) -> Principal {
1841 Principal {
1842 id: principal.id.clone(),
1843 name: principal.name.clone(),
1844 kind: principal_kind_out(principal.kind) as i32,
1845 created_at_unix_ms: principal.created,
1848 }
1849}
1850
1851mod clan_plane;
1853
1854fn key_out(key: &gunnar_core::PublicKey) -> PublicKey {
1855 PublicKey {
1856 fingerprint: key.fingerprint_sha256.clone(),
1857 algo: key.algo.clone(),
1858 principal_id: key.principal_id.clone(),
1859 comment: String::new(),
1862 added_at_unix_ms: key.added,
1863 last_used_unix_ms: key.last_used,
1864 expires_at_unix_ms: key.expires,
1865 }
1866}
1867
1868fn grant_out(grant: &Grant) -> gunnar_proto::Grant {
1873 let (namespace, repo) = match &grant.scope {
1874 Scope::Server => (String::new(), String::new()),
1875 Scope::Namespace(name) => (name.clone(), String::new()),
1876 Scope::Repo(path) => (path.namespace().to_owned(), path.name().to_owned()),
1877 };
1878 gunnar_proto::Grant {
1879 principal_id: grant.principal.clone(),
1880 namespace,
1881 repo,
1882 level: level_out(grant.level) as i32,
1883 granted_at_unix_ms: 0,
1884 }
1885}
1886
1887fn scope_in(namespace: &str, repo: &str) -> ControlResult<Scope> {
1888 match (namespace.is_empty(), repo.is_empty()) {
1889 (true, true) => Ok(Scope::Server),
1890 (false, true) => Ok(Scope::Namespace(namespace.to_owned())),
1891 (false, false) => RepoRef::new(namespace, repo)
1892 .map(Scope::Repo)
1893 .map_err(|err| ControlError::InvalidArgument(err.to_string())),
1894 (true, false) => Err(ControlError::InvalidArgument(
1895 "a repository-scoped grant must name a namespace (multi-tenancy has \
1896 no ungrouped repos)"
1897 .into(),
1898 )),
1899 }
1900}
1901
1902fn object_kind_out(kind: ObjType) -> ControlResult<gunnar_proto::ObjectKind> {
1912 match kind {
1913 ObjType::Blob => Ok(gunnar_proto::ObjectKind::Blob),
1914 ObjType::Tree => Ok(gunnar_proto::ObjectKind::Tree),
1915 ObjType::Commit => Ok(gunnar_proto::ObjectKind::Commit),
1916 ObjType::Tag => Ok(gunnar_proto::ObjectKind::Tag),
1917 other => Err(ControlError::Internal(format!(
1918 "the store answered a read with `{}`, which is a delta; GitServe::read is \
1919 documented to return the resolved type",
1920 other.as_str()
1921 ))),
1922 }
1923}
1924
1925fn oid_at_path(
1938 git: &dyn GitServe,
1939 hash_kind: gix_hash::Kind,
1940 tree: gix_hash::ObjectId,
1941 segments: &[&[u8]],
1942) -> ControlResult<Option<gix_hash::ObjectId>> {
1943 let mut current = tree;
1944 let last = segments.len().saturating_sub(1);
1945 for (index, segment) in segments.iter().enumerate() {
1946 let Some((kind, bytes)) = git.read(current.as_bytes()).map_err(store_error)? else {
1947 return Ok(None);
1948 };
1949 if kind != ObjType::Tree {
1950 return Ok(None);
1951 }
1952 let Some(entry) = object::tree_entry_named(&bytes, hash_kind, segment)
1953 .map_err(|err| ControlError::Internal(format!("decoding tree {current}: {err}")))?
1954 else {
1955 return Ok(None);
1956 };
1957 if index == last {
1958 return Ok(Some(entry.oid));
1959 }
1960 if entry.kind != object::EntryKind::Tree {
1961 return Ok(None);
1962 }
1963 current = entry.oid;
1964 }
1965 Ok(Some(current))
1966}
1967
1968#[derive(Clone, Copy)]
1974enum LogStart<'a> {
1975 One(&'a str),
1976 AllRefs,
1977}
1978
1979fn resolve_log_start(
1982 git: &dyn GitServe,
1983 hash: gix_hash::Kind,
1984 path: &RepoPath,
1985 start: &str,
1986) -> ControlResult<gix_hash::ObjectId> {
1987 let start = if start.is_empty() {
1988 gunnar_store::HEAD
1989 } else {
1990 start
1991 };
1992 let head = match gix_hash::ObjectId::from_hex(start.as_bytes()) {
1993 Ok(id) => id,
1994 Err(_) => {
1995 gunnar_store::refname::validate(start)
1996 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
1997 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
1998 let resolved = if start == gunnar_store::HEAD {
2002 snapshot.head(git).map_err(wire_error)?
2003 } else {
2004 snapshot.resolve(start).map_err(wire_error)?
2005 };
2006 resolved.ok_or_else(|| ControlError::NotFound(format!("{start} in {path}")))?
2007 }
2008 };
2009 if head.kind() != hash {
2010 return Err(ControlError::InvalidArgument(format!(
2011 "{start} is a {} oid; this repository is {hash}",
2012 head.kind()
2013 )));
2014 }
2015 Ok(head)
2016}
2017
2018impl CoreControlPlane {
2019 fn log_walk(
2020 &self,
2021 path: &RepoPath,
2022 start: LogStart<'_>,
2023 limit: u32,
2024 touching: &str,
2025 ) -> ControlResult<Vec<Commit>> {
2026 use std::collections::{BinaryHeap, HashMap, HashSet};
2027
2028 let (_, store) = self.open(path)?;
2029 let git = store.serve();
2030 let hash = store.hash_kind();
2031
2032 let filter: Option<Vec<&[u8]>> = if touching.is_empty() {
2036 None
2037 } else {
2038 let mut segments = Vec::new();
2039 for segment in touching.split('/').filter(|s| !s.is_empty()) {
2040 if segment == "." || segment == ".." {
2041 return Err(ControlError::InvalidArgument(format!(
2042 "the log path filter cannot contain {segment:?}"
2043 )));
2044 }
2045 segments.push(segment.as_bytes());
2046 }
2047 (!segments.is_empty()).then_some(segments)
2048 };
2049
2050 let heads: Vec<gix_hash::ObjectId> = match start {
2051 LogStart::One(start) => vec![resolve_log_start(git, hash, path, start)?],
2052 LogStart::AllRefs => {
2053 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
2054 let mut heads = Vec::new();
2055 if let Some(head) = snapshot.head(git).map_err(wire_error)? {
2056 heads.push(head);
2057 }
2058 for row in snapshot
2059 .rows()
2060 .iter()
2061 .filter(|r| r.name.starts_with("refs/heads/") || r.name.starts_with("refs/tags/"))
2062 {
2063 if let Some(id) = snapshot.resolve_row(row).map_err(wire_error)? {
2064 heads.push(id);
2065 }
2066 }
2067 heads
2068 }
2069 };
2070
2071 let limit = if limit == 0 { 50 } else { limit } as usize;
2077 let mut heap: BinaryHeap<(i64, gix_hash::ObjectId)> = BinaryHeap::new();
2078 let mut seen: HashSet<gix_hash::ObjectId> = HashSet::new();
2079 let mut queued: HashMap<gix_hash::ObjectId, object::Commit> = HashMap::new();
2080 let mut enqueue = |id: gix_hash::ObjectId,
2081 heap: &mut BinaryHeap<(i64, gix_hash::ObjectId)>,
2082 queued: &mut HashMap<gix_hash::ObjectId, object::Commit>|
2083 -> ControlResult<()> {
2084 if !seen.insert(id) {
2085 return Ok(());
2086 }
2087 let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
2090 return Ok(());
2091 };
2092 if kind != ObjType::Commit {
2093 return Ok(());
2094 }
2095 let commit = object::commit(&bytes, hash)
2096 .map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}")))?;
2097 heap.push((commit.committer.seconds, id));
2098 queued.insert(id, commit);
2099 Ok(())
2100 };
2101 for head in heads {
2102 let head = peel(git, hash, head)?;
2103 enqueue(head, &mut heap, &mut queued)?;
2104 }
2105 let mut out = Vec::new();
2106
2107 const MAX_FILTERED_SCAN: usize = 20_000;
2112 let mut scanned = 0usize;
2113 let mut tree_of: HashMap<gix_hash::ObjectId, gix_hash::ObjectId> = HashMap::new();
2116
2117 while let Some((_, id)) = heap.pop() {
2118 if out.len() >= limit {
2119 break;
2120 }
2121 let Some(commit) = queued.remove(&id) else {
2122 continue;
2123 };
2124 let object::Commit {
2125 tree,
2126 parents,
2127 author,
2128 committer,
2129 message,
2130 } = commit;
2131
2132 if let Some(segments) = &filter {
2139 scanned += 1;
2140 if scanned > MAX_FILTERED_SCAN {
2141 break;
2142 }
2143 tree_of.insert(id, tree);
2144 let current = oid_at_path(git, hash, tree, segments)?;
2145 let previous = match parents.first() {
2146 None => None,
2147 Some(parent) => {
2148 let parent_tree = match tree_of.get(parent) {
2149 Some(known) => Some(*known),
2150 None => match git.read(parent.as_bytes()).map_err(store_error)? {
2151 Some((ObjType::Commit, parent_bytes)) => {
2152 let decoded =
2153 object::commit(&parent_bytes, hash).map_err(|err| {
2154 ControlError::Internal(format!(
2155 "decoding commit {parent}: {err}"
2156 ))
2157 })?;
2158 tree_of.insert(*parent, decoded.tree);
2159 Some(decoded.tree)
2160 }
2161 _ => None,
2165 },
2166 };
2167 match parent_tree {
2168 Some(parent_tree) => oid_at_path(git, hash, parent_tree, segments)?,
2169 None => None,
2170 }
2171 }
2172 };
2173 if current == previous {
2174 for parent in parents {
2176 enqueue(parent, &mut heap, &mut queued)?;
2177 }
2178 continue;
2179 }
2180 }
2181
2182 out.push(Commit {
2183 oid: id.to_string(),
2184 parent_oids: parents.iter().map(|p| p.to_string()).collect(),
2185 tree_oid: tree.to_string(),
2186 author: Some(signature_out(
2187 &author.name,
2188 &author.email,
2189 author.seconds,
2190 author.offset_seconds,
2191 )),
2192 committer: Some(signature_out(
2193 &committer.name,
2194 &committer.email,
2195 committer.seconds,
2196 committer.offset_seconds,
2197 )),
2198 message: String::from_utf8_lossy(&message).into_owned(),
2199 });
2200 for parent in parents {
2201 enqueue(parent, &mut heap, &mut queued)?;
2202 }
2203 }
2204 Ok(out)
2205 }
2206}
2207
2208fn peel(
2209 git: &dyn GitServe,
2210 hash_kind: gix_hash::Kind,
2211 mut id: gix_hash::ObjectId,
2212) -> ControlResult<gix_hash::ObjectId> {
2213 for _ in 0..16 {
2214 let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
2215 return Ok(id);
2216 };
2217 if kind != ObjType::Tag {
2218 return Ok(id);
2219 }
2220 id = object::tag_target(&bytes, hash_kind)
2221 .map_err(|err| ControlError::Internal(format!("decoding tag {id}: {err}")))?;
2222 }
2223 Err(ControlError::FailedPrecondition(format!(
2224 "the tag chain at {id} is longer than 16 links, or cyclic"
2225 )))
2226}
2227
2228fn ref_target_out(
2232 git: &dyn GitServe,
2233 hash_kind: gix_hash::Kind,
2234 snapshot: &RefSnapshot,
2235 row: &RefRow,
2236 want_peeled: bool,
2237) -> ControlResult<RefTarget> {
2238 let (oid, symbolic) = match (&row.oid, &row.symref_target) {
2239 (_, Some(inner)) => (snapshot.resolve(inner).map_err(wire_error)?, inner.clone()),
2243 (Some(raw), None) => (
2244 Some(gunnar_wire::refs::oid(raw).map_err(wire_error)?),
2245 String::new(),
2246 ),
2247 (None, None) => (None, String::new()),
2248 };
2249 let peeled = match (want_peeled, oid) {
2250 (true, Some(id)) => {
2255 let peeled = peel(git, hash_kind, id)?;
2256 if peeled == id {
2257 String::new()
2258 } else {
2259 peeled.to_string()
2260 }
2261 }
2262 _ => String::new(),
2263 };
2264 Ok(RefTarget {
2265 ref_name: row.name.clone(),
2266 oid: oid.map(|id| id.to_string()).unwrap_or_default(),
2267 symbolic_target: symbolic,
2268 peeled_oid: peeled,
2269 })
2270}
2271
2272fn signature_out(name: &[u8], email: &[u8], seconds: i64, tz_offset_seconds: i32) -> Signature {
2288 Signature {
2289 name: String::from_utf8_lossy(name).into_owned(),
2290 email: String::from_utf8_lossy(email).into_owned(),
2291 time_unix_seconds: seconds,
2292 tz_offset_minutes: tz_offset_seconds / 60,
2293 }
2294}
2295
2296impl gunnar_grpc::FirstBoot for CoreControlPlane {
2307 fn bootstrap_state(&self) -> ControlResult<gunnar_grpc::BootstrapState> {
2308 let auth = self.stores.auth();
2309 let window_open = auth.bootstrap_window_open().map_err(unavailable)?;
2312 let record = auth.bootstrap_record().map_err(unavailable)?;
2313 let (principals, keys, _grants) = auth.counts().map_err(unavailable)?;
2314 Ok(gunnar_grpc::BootstrapState {
2315 window_open,
2316 bootstrapped: record.is_some(),
2317 bootstrapped_at_unix_ms: record.as_ref().map(|r| r.at).unwrap_or_default(),
2318 bootstrapped_principal: record
2319 .as_ref()
2320 .map(|r| r.principal_id.clone())
2321 .unwrap_or_default(),
2322 bootstrapped_fingerprint: record
2323 .as_ref()
2324 .map(|r| r.fingerprint.clone())
2325 .unwrap_or_default(),
2326 principals: principals as u64,
2327 keys: keys as u64,
2328 serving_git: crate::readiness::is_serving_git(self.git_listener),
2338 readiness: crate::readiness::detail_with(self.git_listener),
2339 build: gunnar_common::BUILD_REF.to_owned(),
2340 role: if self.twin_role || self.stores.is_twin() { "twin" } else { "primary" }.to_owned(),
2341 recharge_from: awaited_twin(&self.stores),
2344 restoring: crate::twin::is_restoring(self.stores.root()),
2345 log_head_seq: self.stores.txlog().head_seq(),
2346 restore_partial: if crate::twin::is_restoring(self.stores.root()) { crate::twin::intake::partial(self.stores.root()) } else { None },
2348 restore_snapshot_resumable: true,
2351 restore_snapshot: if crate::twin::is_restoring(self.stores.root()) {
2352 crate::twin::snapshot::held_snapshot(self.stores.root()).map(|h| (h.id, h.items, h.bytes, h.whole))
2353 } else {
2354 None
2355 },
2356 })
2357 }
2358
2359 fn enroll_first_identity(
2360 &self,
2361 name: &str,
2362 openssh_public_key: &str,
2363 ) -> ControlResult<gunnar_grpc::Enrolled> {
2364 #[cfg(feature = "replication")]
2367 if let Some(twin) = crate::repair::held(&self.stores) {
2368 return Err(ControlError::FailedPrecondition(crate::repair::held_refusal(&twin)));
2369 }
2370 let enrolled = self
2371 .stores
2372 .auth()
2373 .enroll_first_user(name, openssh_public_key)
2374 .map_err(enrolment_error)?;
2375 if let Err(err) = self
2385 .stores
2386 .catalog()
2387 .create_namespace(&enrolled.record.account)
2388 {
2389 tracing::warn!(
2390 account = %enrolled.record.account,
2391 %err,
2392 "the first user was enrolled and the enrolment window is closed, but its account \
2393 namespace could not be created. The user is admin of it by rule and can create \
2394 it themselves with Repos.CreateNamespace."
2395 );
2396 }
2397 let (client_cert_pem, client_key_pem, issuer_cert_pem) = match self.stores.pki() {
2406 Ok(pki) => match pki.issuer() {
2407 Ok(Some(issuer)) => {
2408 match pki.issue_with_server_key(
2409 &enrolled.principal.id,
2410 gunnar_core::DEFAULT_LEAF_DAYS,
2411 ) {
2412 Ok(issued) => {
2413 if let Err(err) = self.stores.auth().register_certificate(
2414 &enrolled.principal.id,
2415 &issued.cert_der,
2416 None,
2417 ) {
2418 tracing::warn!(
2419 principal = %enrolled.principal.id,
2420 %err,
2421 "the bootstrap certificate was minted but could not be \
2422 registered; it is returned anyway and gunnar-admin can \
2423 register it with `gunnar key add`"
2424 );
2425 }
2426 (
2427 issued.cert_pem,
2428 issued.key_pem.unwrap_or_default(),
2429 issuer.cert_pem,
2430 )
2431 }
2432 Err(err) => {
2433 tracing::warn!(
2434 principal = %enrolled.principal.id,
2435 %err,
2436 "no bootstrap certificate: the issuer refused to mint"
2437 );
2438 (String::new(), String::new(), String::new())
2439 }
2440 }
2441 }
2442 Ok(None) => (String::new(), String::new(), String::new()),
2443 Err(err) => {
2444 tracing::warn!(%err, "no bootstrap certificate: the PKI document could not \
2445 be read");
2446 (String::new(), String::new(), String::new())
2447 }
2448 },
2449 Err(_) => (String::new(), String::new(), String::new()),
2450 };
2451 Ok(gunnar_grpc::Enrolled {
2452 principal: principal_out(&enrolled.principal),
2453 key: key_out(&enrolled.key),
2454 account: enrolled.record.account,
2455 bootstrapped_at_unix_ms: enrolled.record.at,
2456 client_cert_pem,
2457 client_key_pem,
2458 issuer_cert_pem,
2459 })
2460 }
2461}
2462
2463fn unavailable(err: gunnar_core::Error) -> ControlError {
2465 ControlError::Unavailable(format!("the identity table could not be read: {err}"))
2466}
2467
2468fn awaited_twin(stores: &ServerStores) -> String {
2474 #[cfg(feature = "wg-client")]
2475 if let Some(twin) = crate::pairing::recharge_awaited(stores) {
2476 return twin;
2477 }
2478 #[cfg(feature = "replication")]
2479 if let Some(twin) = crate::repair::held(stores) {
2480 return twin;
2481 }
2482 let _ = stores;
2483 String::new()
2484}
2485
2486fn enrolment_error(err: gunnar_core::Error) -> ControlError {
2487 match err {
2488 e @ (gunnar_core::Error::AlreadyBootstrapped { .. }
2491 | gunnar_core::Error::AlreadyProvisioned { .. }) => {
2492 ControlError::FailedPrecondition(e.to_string())
2493 }
2494 e @ (gunnar_core::Error::InvalidPath { .. }
2496 | gunnar_core::Error::ReservedPrincipalId(_)
2497 | gunnar_core::Error::Key(_)) => ControlError::InvalidArgument(e.to_string()),
2498 other => ControlError::Unavailable(other.to_string()),
2501 }
2502}
2503
2504impl CoreControlPlane {
2505 pub fn claim_operator_named(
2529 &self,
2530 fed: Option<&str>,
2531 asked: &str,
2532 ) -> ControlResult<gunnar_grpc::OperatorClaim> {
2533 let (name, named_by) = gunnar_grpc::seam::operator_name(fed, asked)?;
2534 let auth = self.stores.auth();
2535 let held = match auth.principal(name).map_err(internal)? {
2536 Some(p) => Some(p),
2537 None => auth.principal_by_name(name).map_err(internal)?,
2538 };
2539 if let Some(p) = held {
2540 if p.kind != PrincipalKind::User {
2541 return Err(ControlError::FailedPrecondition(format!(
2542 "the operator's name {name:?} is held by principal {:?}, which is not a person \
2543 ({:?}); an estate's machines are never its operator",
2544 p.id, p.kind
2545 )));
2546 }
2547 let adopted_namespace = false;
2548 if self.stores.catalog().namespace(&p.id).map_err(internal)?.is_none() {
2549 self.stores.catalog().create_namespace(&p.id).map_err(internal)?;
2550 }
2551 self.grant(&p.id, "", "", GrantLevel::Admin)?;
2552 return Ok(gunnar_grpc::OperatorClaim {
2553 principal: principal_out(&p),
2554 created: false,
2555 adopted_namespace,
2556 named_by,
2557 });
2558 }
2559 let principal =
2560 gunnar_core::Principal::new(name, name, PrincipalKind::User).map_err(internal)?;
2561 let created = auth.add_principal(principal).map_err(internal)?;
2562 let adopted_namespace = match self.stores.catalog().create_namespace(&created.id) {
2563 Ok(_) => false,
2564 Err(gunnar_core::Error::NamespaceExists(_)) => true,
2565 Err(err) => {
2566 let _ = auth.remove_principal(&created.id);
2567 return Err(internal(err));
2568 }
2569 };
2570 if let Err(err) = self.grant(&created.id, "", "", GrantLevel::Admin) {
2571 let _ = auth.remove_principal(&created.id);
2574 return Err(err);
2575 }
2576 self.stores.mirror_identity();
2577 tracing::warn!(
2578 target: "gunnar::audit",
2579 event = "operator.claimed",
2580 operator = %created.id,
2581 adopted_namespace,
2582 named_by,
2583 "the estate's operator was created: a person, keyless, gunnar-admin, with its account \
2584 namespace{}",
2585 if adopted_namespace {
2586 " — the namespace already existed with no owner (a first push conjured it) and is \
2587 now the operator's account"
2588 } else {
2589 ""
2590 }
2591 );
2592 Ok(gunnar_grpc::OperatorClaim {
2593 principal: principal_out(&created),
2594 created: true,
2595 adopted_namespace,
2596 named_by,
2597 })
2598 }
2599}
2600
2601mod enterprise_plane;
2603
2604impl ControlPlane for CoreControlPlane {
2605 fn enterprise(&self) -> Option<&dyn gunnar_grpc::enterprise::EnterprisePlane> {
2606 Some(self)
2607 }
2608
2609 fn create_namespace(&self, name: &str, _description: &str) -> ControlResult<Namespace> {
2612 if self.stores.auth().employer_of(name).map_err(internal)?.is_some() {
2616 return Err(ControlError::AlreadyExists(format!(
2617 "{name} is an employee's id, and an employee holds no account of its own"
2618 )));
2619 }
2620 let namespace = self
2621 .stores
2622 .catalog()
2623 .create_namespace(name)
2624 .map_err(internal)?;
2625 Ok(self.namespace_out(namespace, 0))
2626 }
2627
2628 fn list_namespaces(&self) -> ControlResult<Vec<Namespace>> {
2629 let mut employees: std::collections::BTreeMap<String, u32> = Default::default();
2630 for principal in self.stores.auth().principals().map_err(internal)? {
2631 if let Some(account) = self.stores.auth().employer_of(&principal.id).map_err(internal)? {
2632 *employees.entry(account).or_default() += 1;
2633 }
2634 }
2635 Ok(self
2636 .stores
2637 .catalog()
2638 .namespaces()
2639 .map_err(internal)?
2640 .into_iter()
2641 .map(|ns| {
2642 let count = employees.get(&ns.name).copied().unwrap_or(0);
2643 self.namespace_out(ns, count)
2644 })
2645 .collect())
2646 }
2647
2648 fn namespace(&self, name: &str) -> ControlResult<Option<Namespace>> {
2651 let Some(ns) = self.stores.catalog().namespace(name).map_err(internal)? else {
2652 return Ok(None);
2653 };
2654 let count = self.employee_count(name)?;
2655 Ok(Some(self.namespace_out(ns, count)))
2656 }
2657
2658 fn namespace_for_outsider(&self, principal_id: &str, name: &str) -> ControlResult<Option<Namespace>> {
2659 let auth = self.stores.auth();
2662 if auth.gate_covering(principal_id).map_err(internal)?.is_some()
2663 || auth.gate_covering(name).map_err(internal)?.is_some()
2664 {
2665 return Ok(None);
2666 }
2667 self.namespace(name)
2668 }
2669
2670 fn set_namespace_settings(
2671 &self,
2672 name: &str,
2673 public: Option<bool>,
2674 events_public: Option<bool>,
2675 ) -> ControlResult<Namespace> {
2676 let catalog = self.stores.catalog();
2677 let mut ns = catalog
2678 .namespace(name)
2679 .map_err(internal)?
2680 .ok_or_else(|| ControlError::NotFound(format!("namespace {name}")))?;
2681 if let Some(public) = public {
2682 ns = catalog.set_namespace_public(name, public).map_err(internal)?;
2683 }
2684 if let Some(events_public) = events_public {
2685 ns = catalog
2686 .set_namespace_events_public(name, events_public)
2687 .map_err(internal)?;
2688 }
2689 let count = self.employee_count(name)?;
2690 Ok(self.namespace_out(ns, count))
2691 }
2692
2693 fn delete_namespace(&self, name: &str, delete_repos: bool) -> ControlResult<u32> {
2694 if self.stores.auth().is_clan(name).map_err(internal)? {
2699 return Err(ControlError::FailedPrecondition(format!(
2700 "namespace {name} is clan {name}'s: it goes when the clan's admin dissolves the clan"
2701 )));
2702 }
2703 let gone = self
2704 .stores
2705 .catalog()
2706 .delete_namespace(name, delete_repos)
2707 .map_err(internal)?;
2708 if let Err(err) = self.stores.flow().forget(name) {
2719 tracing::warn!(
2720 target: "gunnar::store",
2721 event = "flow.forget_failed",
2722 namespace = %name,
2723 %err,
2724 "the namespace was deleted but its activity series could not be removed"
2725 );
2726 }
2727 Ok(gone.len() as u32)
2728 }
2729
2730 fn create_repo(&self, spec: ProtoNewRepo) -> ControlResult<Repo> {
2733 let path = Self::repo_ref(&spec.path)?;
2734 if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
2740 return Err(ControlError::ResourceExhausted(line));
2741 }
2742 let mut new = gunnar_core::NewRepo::new(path).visibility(visibility_in(spec.visibility)?);
2743 if spec.hash_kind != gunnar_proto::HashKind::Unspecified {
2744 new = new.hash_kind(match spec.hash_kind {
2745 gunnar_proto::HashKind::Sha1 => HashKind::Sha1,
2746 _ => HashKind::Sha256,
2747 });
2748 }
2749 if !spec.default_branch.is_empty() {
2750 new = new.default_branch(spec.default_branch.clone());
2751 }
2752 let repo = self
2761 .stores
2762 .create_repo(new, spec.adopt)
2763 .map_err(|err| create_error(&err, &spec.path))?;
2764 Ok(repo_out(&repo))
2765 }
2766
2767 fn write_first_commit(&self, path: &RepoPath, readme: &str) -> ControlResult<String> {
2768 let (repo, store) = self.open(path)?;
2769 let kind = match repo.hash_kind {
2770 HashKind::Sha1 => gix_hash::Kind::Sha1,
2771 _ => gix_hash::Kind::Sha256,
2772 };
2773 let when = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or(0);
2774 let (pack, commit) = crate::first_commit::first_commit_pack(kind, "README.md", readme.as_bytes(), "Welcome", when)
2775 .map_err(ControlError::InvalidArgument)?;
2776 store
2777 .serve()
2778 .put(&pack, &[gunnar_store::git_storage_trait::RefUpdate::set(repo.default_branch.clone(), commit.clone())])
2779 .map_err(|err| ControlError::Internal(format!("writing the first commit of {path}: {err:#}")))?;
2780 Ok(commit)
2781 }
2782
2783 fn create_room(&self, namespace: &str, machine: bool) -> ControlResult<()> {
2788 use gunnar_ssh::{PlanAction, PlanVerdict};
2789 for gate in self.stores.plan_gates().iter().filter(|gate| !machine || gate.applies_to_machine()) {
2790 if let PlanVerdict::Refuse(line) = gate.verdict(namespace, PlanAction::Write) {
2791 return Err(ControlError::FailedPrecondition(line));
2792 }
2793 if let PlanVerdict::Refuse(line) = gate.pack_quota(namespace) {
2794 return Err(ControlError::ResourceExhausted(line));
2795 }
2796 }
2797 Ok(())
2798 }
2799
2800 fn get_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
2801 let repo_ref = Self::repo_ref(path)?;
2802 self.stores
2803 .catalog()
2804 .get(&repo_ref)
2805 .map_err(internal)?
2806 .as_ref()
2807 .map(repo_out)
2808 .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))
2809 }
2810
2811 fn list_repos(&self, namespace: Option<&str>) -> ControlResult<Vec<Repo>> {
2812 Ok(self
2813 .stores
2814 .catalog()
2815 .list(namespace)
2816 .map_err(internal)?
2817 .iter()
2818 .map(repo_out)
2819 .collect())
2820 }
2821
2822 fn delete_repo(
2843 &self,
2844 path: &RepoPath,
2845 options: gunnar_grpc::seam::DeleteOptions,
2846 ) -> ControlResult<gunnar_grpc::seam::DeleteOutcome> {
2847 let repo_ref = Self::repo_ref(path)?;
2848 let was = self.stores.catalog().get(&repo_ref).map_err(internal)?;
2852 let orphan = was.is_none() && self.stores.is_orphan(&repo_ref).unwrap_or(false);
2861 if orphan && !options.purge {
2862 return Err(ControlError::FailedPrecondition(
2863 crate::stores::ServerStores::orphan_soft_delete_refusal(&repo_ref),
2864 ));
2865 }
2866 match self.stores.catalog().delete(&repo_ref) {
2870 Ok(_) => {}
2871 Err(gunnar_core::Error::NoSuchRepo(_))
2872 if orphan || self.stores.is_tombstoned(&repo_ref).unwrap_or(false) => {}
2873 Err(err) => return Err(internal(err)),
2874 }
2875 if options.purge {
2876 let (before, after) = self.stores.purge_repo(&repo_ref).map_err(|err| {
2877 ControlError::Internal(format!("removing the repository's directory: {err:#}"))
2878 })?;
2879 return Ok(gunnar_grpc::seam::DeleteOutcome::Purged {
2880 bytes_reclaimed: before.saturating_sub(after),
2881 });
2882 }
2883 let retention = std::time::Duration::from_secs(
2884 u64::from(
2885 options
2886 .retention_days
2887 .unwrap_or(crate::stores::DEFAULT_DELETE_RETENTION_DAYS),
2888 ) * 86_400,
2889 );
2890 let record = self
2891 .stores
2892 .soft_delete(&repo_ref, was.as_ref(), &options.principal_id, retention)
2893 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
2894 let bytes_kept = self
2895 .stores
2896 .repo_dir(&repo_ref)
2897 .map(|dir| crate::stores::dir_bytes(&dir))
2898 .unwrap_or(0);
2899 Ok(gunnar_grpc::seam::DeleteOutcome::Tombstoned {
2900 purge_after_unix_ms: record.purge_after_unix_ms.unwrap_or_default(),
2901 bytes_kept,
2902 })
2903 }
2904
2905 fn undelete_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
2914 let repo_ref = Self::repo_ref(path)?;
2915 if self
2916 .stores
2917 .catalog()
2918 .get(&repo_ref)
2919 .map_err(internal)?
2920 .is_some()
2921 {
2922 return Err(ControlError::AlreadyExists(format!(
2923 "{path} is in the catalog already; it is not deleted"
2924 )));
2925 }
2926 let spec = self
2927 .stores
2928 .undelete_spec(&repo_ref)
2929 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
2930 let repo = self
2936 .stores
2937 .create_repo(spec, true)
2938 .map_err(|err| ControlError::Internal(format!("restoring {path}: {err:#}")))?;
2939 self.stores
2940 .clear_tombstone(&repo_ref)
2941 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
2942 Ok(repo_out(&repo))
2943 }
2944
2945 fn purge_repo(&self, path: &RepoPath) -> ControlResult<u64> {
2946 let repo_ref = Self::repo_ref(path)?;
2947 if self
2948 .stores
2949 .catalog()
2950 .get(&repo_ref)
2951 .map_err(internal)?
2952 .is_some()
2953 {
2954 return Err(ControlError::FailedPrecondition(format!(
2955 "{path} is a LIVE repository, not a deleted one. `repo delete {path} --purge` \
2956 deletes and reclaims it in one step; purge alone only ends a retention window."
2957 )));
2958 }
2959 let tombstone = self
2960 .stores
2961 .tombstone_of(&repo_ref)
2962 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
2963 if tombstone.is_none() {
2964 return Err(ControlError::NotFound(format!(
2965 "{path} is neither catalogued nor tombstoned; there is nothing to purge"
2966 )));
2967 }
2968 let (before, after) = self
2969 .stores
2970 .purge_repo(&repo_ref)
2971 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
2972 Ok(before.saturating_sub(after))
2973 }
2974
2975 fn gc_repo(&self, path: &RepoPath) -> ControlResult<gunnar_grpc::seam::GcJob> {
2987 let repo_ref = Self::repo_ref(path)?;
2988 if self
2995 .stores
2996 .catalog()
2997 .get(&repo_ref)
2998 .map_err(internal)?
2999 .is_none()
3000 {
3001 return Err(ControlError::NotFound(format!(
3002 "{path} is not a live repository; nothing to gc"
3003 )));
3004 }
3005 Ok(gc_job_out(path, self.gc.enqueue(&repo_ref.to_string())))
3006 }
3007
3008 fn gc_status(&self, path: &RepoPath) -> ControlResult<Option<gunnar_grpc::seam::GcJob>> {
3009 let repo_ref = Self::repo_ref(path)?;
3010 Ok(self
3011 .gc
3012 .status(&repo_ref.to_string())
3013 .map(|job| gc_job_out(path, job)))
3014 }
3015
3016 fn gate_account(
3019 &self,
3020 account: &str,
3021 actor: &str,
3022 note: &str,
3023 ) -> ControlResult<(gunnar_grpc::seam::AccountGate, bool)> {
3024 let before = self
3031 .stores
3032 .auth()
3033 .gate_covering(account)
3034 .map_err(internal)?
3035 .is_some();
3036 let gate = self
3037 .stores
3038 .auth()
3039 .gate(gunnar_core::AccountGate::administrative(account, actor, note))
3040 .map_err(gate_error)?;
3041 tracing::warn!(
3042 target: "gunnar::audit",
3043 event = "account.gated",
3044 account = %account,
3045 actor = %actor,
3046 note = %note,
3047 already = before,
3048 "an account was locked out: every level, both doors, until it is ungated"
3049 );
3050 Ok((gate_out(gate), before))
3051 }
3052
3053 fn ungate_account(
3054 &self,
3055 account: &str,
3056 ) -> ControlResult<Option<gunnar_grpc::seam::AccountGate>> {
3057 let lifted = self
3058 .stores
3059 .auth()
3060 .lift_gate(account)
3061 .map_err(internal)?
3062 .map(gate_out);
3063 if let Some(gate) = &lifted {
3064 tracing::warn!(
3065 target: "gunnar::audit",
3066 event = "account.ungated",
3067 account = %account,
3068 held_by = %gate.holder,
3069 "an account lockout was lifted"
3070 );
3071 }
3072 Ok(lifted)
3073 }
3074
3075 fn account_gates(&self) -> ControlResult<Vec<gunnar_grpc::seam::AccountGate>> {
3076 Ok(self
3077 .stores
3078 .auth()
3079 .gates()
3080 .map_err(internal)?
3081 .into_iter()
3082 .map(gate_out)
3083 .collect())
3084 }
3085
3086 fn purge_account(
3104 &self,
3105 account: &str,
3106 options: gunnar_grpc::seam::PurgeAccountOptions,
3107 ) -> ControlResult<(gunnar_grpc::seam::PurgeJob, gunnar_grpc::seam::AccountGate)> {
3108 let account = account.trim();
3109 if account.is_empty() {
3110 return Err(ControlError::InvalidArgument(
3111 "an account purge needs an account name".into(),
3112 ));
3113 }
3114 if let Some(p) = self.stores.auth().principal(account).map_err(internal)? {
3121 if p.kind != gunnar_core::PrincipalKind::User {
3122 return Err(ControlError::FailedPrecondition(format!(
3123 "{account} is a machine principal ({:?}), not a person: a machine's account is never \
3124 purged. Remove the principal itself if it is really to go.",
3125 p.kind
3126 )));
3127 }
3128 }
3129 let job_id = self.purge.next_job_id();
3130 self.gate_for_purge(account, &job_id, &options)?;
3135
3136 let order = crate::purge_queue::PurgeOrder::new(account, &job_id, &options.reason)
3137 .by(&options.principal_id)
3138 .removing_principal(options.remove_principal)
3139 .expecting_empty(options.expect_empty);
3140 let job = self.purge.enqueue(order);
3141
3142 let gate = self
3143 .stores
3144 .auth()
3145 .gate_covering(account)
3146 .map_err(internal)?
3147 .map(gate_out)
3148 .unwrap_or_else(|| gunnar_grpc::seam::AccountGate {
3153 account: account.to_owned(),
3154 reason: gunnar_grpc::seam::GateReason::Purge,
3155 holder: job_id.clone(),
3156 actor: options.principal_id.clone(),
3157 note: "the lockout was lifted between being taken and being read back".into(),
3158 at_unix_ms: gunnar_core::now_ms(),
3159 expires_unix_ms: 0,
3160 });
3161 Ok((purge_job_out(job), gate))
3162 }
3163
3164 fn purge_account_status(
3165 &self,
3166 account: &str,
3167 ) -> ControlResult<(
3168 Option<gunnar_grpc::seam::PurgeJob>,
3169 Option<gunnar_grpc::seam::AccountGate>,
3170 )> {
3171 let job = self.purge.status(account).map(purge_job_out);
3172 let gate = self
3173 .stores
3174 .auth()
3175 .gate_covering(account)
3176 .map_err(internal)?
3177 .map(gate_out);
3178 Ok((job, gate))
3179 }
3180
3181 fn list_deleted(
3182 &self,
3183 namespace: Option<&str>,
3184 ) -> ControlResult<Vec<gunnar_grpc::seam::DeletedRepoRow>> {
3185 Ok(self
3186 .stores
3187 .tombstones()
3188 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
3189 .into_iter()
3190 .filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
3191 .map(|row| gunnar_grpc::seam::DeletedRepoRow {
3192 path: RepoPath {
3193 namespace: row.path.namespace().to_owned(),
3194 repo: row.path.name().to_owned(),
3195 },
3196 deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
3197 purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
3198 principal_id: row.tombstone.principal_id,
3199 bytes: row.bytes,
3200 })
3201 .collect())
3202 }
3203
3204 fn list_orphans(
3214 &self,
3215 namespace: Option<&str>,
3216 ) -> ControlResult<Vec<gunnar_grpc::seam::OrphanRow>> {
3217 Ok(self
3218 .stores
3219 .orphans()
3220 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
3221 .into_iter()
3222 .filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
3223 .map(|row| gunnar_grpc::seam::OrphanRow {
3224 path: RepoPath {
3225 namespace: row.path.namespace().to_owned(),
3226 repo: row.path.name().to_owned(),
3227 },
3228 dir: row.dir.display().to_string(),
3229 bytes: row.bytes,
3230 hash_marker: row.hash_marker.unwrap_or_default(),
3231 modified_unix_ms: row.modified_unix_ms,
3232 failed_reclaim: row.failed_reclaim.unwrap_or_default(),
3233 })
3234 .collect())
3235 }
3236
3237 fn reclaim_orphan(&self, path: &RepoPath) -> ControlResult<u64> {
3241 let repo_ref = Self::repo_ref(path)?;
3242 self.stores
3243 .reclaim_orphan(&repo_ref)
3244 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
3245 }
3246
3247 fn index_progress(&self) -> ControlResult<Vec<gunnar_grpc::seam::IndexProgressRow>> {
3248 Ok(self
3249 .stores
3250 .engine()
3251 .index_progress()
3252 .into_iter()
3253 .filter_map(|(dir, s)| {
3254 let (namespace, name) = self.stores.repo_of_dir(&dir)?;
3255 Some(gunnar_grpc::seam::IndexProgressRow {
3256 namespace,
3257 name,
3258 pack_id: s.pack_id,
3259 objects_done: s.objects_done,
3260 objects_total: s.objects_total,
3261 bytes_done: s.bytes_done,
3262 bytes_total: s.bytes_total,
3263 phase: if s.active {
3264 s.phase.to_string()
3265 } else {
3266 "idle".to_string()
3267 },
3268 started_at_unix_ms: s.started_at_ms as i64,
3269 resumed: s.resumed,
3270 reinflations: s.reinflations,
3271 pauses: s.pauses,
3272 })
3273 })
3274 .filter(|r| r.phase != "idle")
3275 .collect())
3276 }
3277
3278 fn set_default_branch(&self, path: &RepoPath, default_branch: &str) -> ControlResult<Repo> {
3279 let repo_ref = Self::repo_ref(path)?;
3280 let repo = self
3281 .stores
3282 .catalog()
3283 .set_default_branch(&repo_ref, default_branch)
3284 .map_err(internal)?;
3285 self.stores.repoint_head(&repo).map_err(|err| {
3293 ControlError::Internal(format!("pointing HEAD at the new default branch: {err:#}"))
3294 })?;
3295 Ok(repo_out(&repo))
3296 }
3297
3298 fn set_visibility(
3299 &self,
3300 path: &RepoPath,
3301 visibility: gunnar_proto::Visibility,
3302 ) -> ControlResult<Repo> {
3303 if visibility == gunnar_proto::Visibility::Unspecified {
3304 return Err(ControlError::InvalidArgument(
3305 "visibility unspecified".into(),
3306 ));
3307 }
3308 let repo_ref = Self::repo_ref(path)?;
3309 self.stores
3310 .catalog()
3311 .set_visibility(&repo_ref, visibility_in(visibility)?)
3312 .map(|repo| repo_out(&repo))
3313 .map_err(internal)
3314 }
3315
3316 fn recipient_keys(&self, principal_id: &str) -> ControlResult<Vec<gunnar_proto::RecipientKey>> {
3320 let public = self.stores.encryption_keys().public_of(principal_id).map_err(key_refusal)?;
3321 Ok(public
3322 .into_iter()
3323 .map(|public| gunnar_proto::RecipientKey {
3324 fingerprint: gunnar_core::auth::fingerprint_of_blob(public.as_bytes()).to_string(),
3325 algo: if public.starts_with("age1pq1") { gunnar_core::auth::PQ_RECIPIENT_ALGO.to_owned() } else { "X25519".to_owned() },
3326 openssh: public,
3327 })
3328 .collect())
3329 }
3330
3331 fn encryption_key(&self, principal_id: &str) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3332 Ok(encryption_key_out(self.stores.encryption_keys().get(principal_id).map_err(key_refusal)?))
3333 }
3334
3335 fn create_encryption_key(
3336 &self,
3337 principal_id: &str,
3338 public_key: &str,
3339 first: gunnar_proto::UnlockMethod,
3340 ) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3341 let first = method_in(first)?;
3342 let key = self.stores.encryption_keys().create(principal_id, public_key, first, gunnar_core::now_ms().max(0) as u64).map_err(key_refusal)?;
3343 Ok(encryption_key_out(Some(key)))
3344 }
3345
3346 fn add_unlock_method(&self, principal_id: &str, method: gunnar_proto::UnlockMethod) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3347 let method = method_in(method)?;
3348 let key = self.stores.encryption_keys().add_method(principal_id, method, gunnar_core::now_ms().max(0) as u64).map_err(key_refusal)?;
3349 Ok(encryption_key_out(Some(key)))
3350 }
3351
3352 fn remove_unlock_method(&self, principal_id: &str, id: &str) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3353 let key = self.stores.encryption_keys().remove_method(principal_id, id).map_err(key_refusal)?;
3354 Ok(encryption_key_out(Some(key)))
3355 }
3356
3357 fn set_encrypted(&self, path: &RepoPath, encrypted: bool) -> ControlResult<Repo> {
3358 let repo_ref = Self::repo_ref(path)?;
3359 self.stores
3360 .catalog()
3361 .set_encrypted(&repo_ref, encrypted)
3362 .map(|repo| repo_out(&repo))
3363 .map_err(internal)
3364 }
3365
3366 #[cfg(feature = "engine-znippy")]
3377 fn archive_out(
3378 &self,
3379 path: &RepoPath,
3380 from_generation: u64,
3381 sink: &mut dyn std::io::Write,
3382 ) -> ControlResult<gunnar_grpc::seam::ArchiveCut> {
3383 let repo_ref = Self::repo_ref(path)?;
3384 let dir = self
3385 .stores
3386 .repo_dir(&repo_ref)
3387 .map_err(|err| ControlError::Internal(format!("resolving {path}: {err:#}")))?;
3388 if !dir.is_dir() {
3389 return Err(ControlError::NotFound(format!(
3390 "{path} has nothing on disk at {}",
3391 dir.display()
3392 )));
3393 }
3394 let receipt =
3395 crate::storage::znippy_engine::archive_out_to(&dir, &mut { sink }, from_generation)
3396 .map_err(archive_error)?;
3397 self.stores.flow().record(
3410 &path.namespace,
3411 gunnar_core::now_ms(),
3412 &gunnar_core::Transfer {
3413 bytes_out: receipt.bytes_out,
3414 fetched: true,
3415 ..gunnar_core::Transfer::default()
3416 },
3417 );
3418 Ok(gunnar_grpc::seam::ArchiveCut {
3419 from_generation: receipt.from_generation,
3420 to_generation: receipt.to_generation,
3421 bytes_out: receipt.bytes_out,
3422 file_sha256: receipt.file_sha256,
3423 })
3424 }
3425
3426 fn list_refs(
3429 &self,
3430 path: &RepoPath,
3431 prefix: &str,
3432 peel_them: bool,
3433 ) -> ControlResult<Vec<RefTarget>> {
3434 let (_, store) = self.open(path)?;
3435 let git = store.serve();
3436 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3441 let mut out = Vec::new();
3442 for row in snapshot.with_prefix(Some(prefix)) {
3443 out.push(ref_target_out(
3444 git,
3445 store.hash_kind(),
3446 &snapshot,
3447 row,
3448 peel_them,
3449 )?);
3450 }
3451 Ok(out)
3452 }
3453
3454 fn get_ref(
3455 &self,
3456 path: &RepoPath,
3457 ref_name: &str,
3458 peel_them: bool,
3459 ) -> ControlResult<RefTarget> {
3460 let (_, store) = self.open(path)?;
3461 let git = store.serve();
3462 gunnar_store::refname::validate(ref_name)
3468 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
3469 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3470 let head_row;
3476 let row = if ref_name == gunnar_store::HEAD {
3477 head_row = git
3478 .head()
3479 .map_err(store_error)?
3480 .ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?;
3481 &head_row
3482 } else {
3483 snapshot
3484 .get(ref_name)
3485 .ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?
3486 };
3487 ref_target_out(git, store.hash_kind(), &snapshot, row, peel_them)
3488 }
3489
3490 fn log(
3491 &self,
3492 path: &RepoPath,
3493 start: &str,
3494 limit: u32,
3495 touching: &str,
3496 ) -> ControlResult<Vec<Commit>> {
3497 self.log_walk(path, LogStart::One(start), limit, touching)
3498 }
3499
3500 fn log_all_refs(
3504 &self,
3505 path: &RepoPath,
3506 limit: u32,
3507 touching: &str,
3508 ) -> ControlResult<Vec<Commit>> {
3509 self.log_walk(path, LogStart::AllRefs, limit, touching)
3510 }
3511
3512 fn create_pull(
3513 &self,
3514 path: &RepoPath,
3515 author: &str,
3516 title: &str,
3517 description: &str,
3518 source_ref: &str,
3519 target_ref: &str,
3520 ) -> ControlResult<PullRequest> {
3521 for (what, name) in [("source_ref", source_ref), ("target_ref", target_ref)] {
3522 if !name.starts_with("refs/heads/") {
3523 return Err(ControlError::InvalidArgument(format!(
3524 "{what} must be a full branch name under refs/heads/, got {name:?}: a pull \
3525 request merges branches, and a short name stored today is an ambiguity \
3526 resolved differently tomorrow"
3527 )));
3528 }
3529 }
3530 if source_ref == target_ref {
3531 return Err(ControlError::InvalidArgument(
3532 "a branch cannot be merged into itself".to_owned(),
3533 ));
3534 }
3535 if title.trim().is_empty() {
3536 return Err(ControlError::InvalidArgument(
3537 "a title is required".to_owned(),
3538 ));
3539 }
3540 let repo_ref = Self::repo_ref(path)?;
3541 let (_, store) = self.open(path)?;
3542 let git = store.serve();
3543 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3544 let source = snapshot
3545 .resolve(source_ref)
3546 .map_err(wire_error)?
3547 .ok_or_else(|| ControlError::NotFound(format!("{source_ref} in {path}")))?;
3548 snapshot
3549 .resolve(target_ref)
3550 .map_err(wire_error)?
3551 .ok_or_else(|| ControlError::NotFound(format!("{target_ref} in {path}")))?;
3552 let record = self
3553 .pulls()?
3554 .create(
3555 &repo_ref.to_string(),
3556 author,
3557 title,
3558 description,
3559 source_ref,
3560 target_ref,
3561 &source.to_string(),
3562 )
3563 .map_err(internal)?;
3564 Ok(pull_out(path, record))
3565 }
3566
3567 fn list_pulls(&self, path: &RepoPath, state: i32) -> ControlResult<Vec<PullRequest>> {
3568 let repo_ref = Self::repo_ref(path)?;
3569 self.get_repo(path)?;
3572 let filter = match PullRequestState::try_from(state) {
3573 Ok(PullRequestState::Unspecified) | Err(_) => None,
3574 Ok(PullRequestState::Open) => Some(gunnar_core::PullState::Open),
3575 Ok(PullRequestState::Merged) => Some(gunnar_core::PullState::Merged),
3576 Ok(PullRequestState::Closed) => Some(gunnar_core::PullState::Closed),
3577 };
3578 Ok(self
3579 .pulls()?
3580 .list(&repo_ref.to_string(), filter)
3581 .map_err(internal)?
3582 .into_iter()
3583 .map(|record| pull_out(path, record))
3584 .collect())
3585 }
3586
3587 fn get_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
3588 let repo_ref = Self::repo_ref(path)?;
3589 self.get_repo(path)?;
3590 self.pulls()?
3591 .get(&repo_ref.to_string(), number)
3592 .map_err(internal)?
3593 .map(|record| pull_out(path, record))
3594 .ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))
3595 }
3596
3597 fn close_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
3598 let repo_ref = Self::repo_ref(path)?;
3599 self.get_repo(path)?;
3600 match self
3601 .pulls()?
3602 .close(&repo_ref.to_string(), number)
3603 .map_err(internal)?
3604 {
3605 gunnar_core::PullTransition::Done(record) => Ok(pull_out(path, record)),
3606 gunnar_core::PullTransition::NotFound => Err(ControlError::NotFound(format!(
3607 "pull request #{number} in {path}"
3608 ))),
3609 gunnar_core::PullTransition::NotOpen(state) => Err(ControlError::FailedPrecondition(
3610 format!("pull request #{number} is already {state:?}, not open"),
3611 )),
3612 }
3613 }
3614
3615 fn merge_pull(
3623 &self,
3624 path: &RepoPath,
3625 number: u64,
3626 principal_id: &str,
3627 ) -> ControlResult<(PullRequest, MergedRef)> {
3628 let repo_ref = Self::repo_ref(path)?;
3629 let key = repo_ref.to_string();
3630 let pulls = self.pulls()?;
3631 let record = pulls
3632 .get(&key, number)
3633 .map_err(internal)?
3634 .ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))?;
3635 if record.state != gunnar_core::PullState::Open {
3636 return Err(ControlError::FailedPrecondition(format!(
3637 "pull request #{number} is already {:?}, not open",
3638 record.state
3639 )));
3640 }
3641
3642 let (repo_record, store) = self.open(path)?;
3643 let git = store.serve();
3644 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3645 let source = snapshot
3646 .resolve(&record.source_ref)
3647 .map_err(wire_error)?
3648 .ok_or_else(|| {
3649 ControlError::FailedPrecondition(format!(
3650 "the source branch {} is gone; nothing to merge",
3651 record.source_ref
3652 ))
3653 })?;
3654 let target = snapshot
3655 .resolve(&record.target_ref)
3656 .map_err(wire_error)?
3657 .ok_or_else(|| {
3658 ControlError::FailedPrecondition(format!(
3659 "the target branch {} is gone",
3660 record.target_ref
3661 ))
3662 })?;
3663
3664 let head = source.to_string();
3680 let signed = self.forge_events()?;
3683 let events: Vec<gunnar_core::ForgeEvent> = signed
3684 .iter()
3685 .filter(|s| s.event.repo == key)
3686 .map(|s| s.event.clone())
3687 .collect();
3688 let review = gunnar_core::forge::fold(number, &events);
3689 let approved = self.review_state(&key, &review, &signed)?;
3690 let ok = approved
3691 .approvals
3692 .iter()
3693 .any(|a| a.verified && a.commit == head);
3694 if !ok {
3695 let named: Vec<String> = approved
3696 .approvals
3697 .iter()
3698 .map(|a| {
3699 format!(
3700 "{} → {} [{}]",
3701 a.actor,
3702 &a.commit[..12.min(a.commit.len())],
3703 a.verdict
3704 )
3705 })
3706 .collect();
3707 return Err(ControlError::FailedPrecondition(format!(
3708 "no verified approval names {}, the commit {} is at. gunnar does not merge on \
3709 an approval of something else: an approval names the commit it approved, so a \
3710 branch that moved after the review has not been reviewed.{}",
3711 &head[..12.min(head.len())],
3712 record.source_ref,
3713 if named.is_empty() {
3714 " Nobody has approved this pull request.".to_owned()
3715 } else {
3716 format!(" What is approved: {}", named.join(", "))
3717 }
3718 )));
3719 }
3720
3721 let policy = self.policy_of(&repo_record)?;
3734 let rules: Vec<&gunnar_policy::ProtectionRule> = policy
3735 .protection
3736 .iter()
3737 .filter(|rule| rule.pattern.matches_ref(&record.target_ref))
3738 .collect();
3739 let forbid_self = rules.iter().any(|rule| rule.forbid_self_approval);
3740 let required = rules
3741 .iter()
3742 .map(|rule| rule.required_approvals)
3743 .max()
3744 .unwrap_or(0)
3745 .max(1) as usize;
3746 let approvers: Vec<&str> = approved
3747 .approvals
3748 .iter()
3749 .filter(|a| a.verified && a.commit == head)
3750 .map(|a| a.actor.as_str())
3751 .collect();
3752 let counted: Vec<&str> = approvers
3753 .iter()
3754 .copied()
3755 .filter(|actor| !(forbid_self && *actor == record.author))
3756 .collect();
3757 if forbid_self && counted.is_empty() && !approvers.is_empty() {
3758 return Err(ControlError::FailedPrecondition(format!(
3759 "the protection on {} forbids self-approval, and the only verified approval \
3760 of {} is the author's own ({}). An author's sign-off on their own work is \
3761 not a review; somebody else has to look at it.",
3762 record.target_ref,
3763 &head[..12.min(head.len())],
3764 record.author
3765 )));
3766 }
3767 if counted.len() < required {
3768 return Err(ControlError::FailedPrecondition(format!(
3769 "the protection on {} requires {required} distinct verified approvals of {} \
3770 and there {} ({}). Approvals by the same principal count once{}.",
3771 record.target_ref,
3772 &head[..12.min(head.len())],
3773 if counted.len() == 1 {
3774 "is 1".to_owned()
3775 } else {
3776 format!("are {}", counted.len())
3777 },
3778 if counted.is_empty() {
3779 "nobody".to_owned()
3780 } else {
3781 counted.join(", ")
3782 },
3783 if forbid_self {
3784 ", and the author's own does not count here"
3785 } else {
3786 ""
3787 }
3788 )));
3789 }
3790
3791 if source != target {
3810 let principal = self
3811 .stores
3812 .auth()
3813 .principal(principal_id)
3814 .map_err(internal)?
3815 .ok_or_else(|| {
3816 ControlError::FailedPrecondition(format!(
3821 "the principal {principal_id} is no longer in the identity table; \
3822 refusing to move a protected reference for an identity that \
3823 cannot be resolved"
3824 ))
3825 })?;
3826 let gate = gunnar_policy::BranchProtectionGate::new(policy.protection.clone());
3827 let repo_id = gunnar_policy::RepoId::new(&path.namespace, &path.repo);
3828 let policy_principal = gunnar_policy::Principal {
3829 id: principal.id.clone(),
3830 name: principal.name.clone(),
3831 kind: crate::transport::policy_principal_kind(principal.kind),
3832 };
3833 let introduced = gunnar_policy::ancestry::commits_between(
3840 git,
3841 &source,
3842 &target,
3843 gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET,
3844 )
3845 .map_err(|err| {
3846 ControlError::FailedPrecondition(format!(
3847 "the commits this merge would add could not be enumerated ({err}); \
3848 refusing rather than running branch protection over an empty set"
3849 ))
3850 })?;
3851 let updates = [gunnar_policy::RefUpdate::from_wire(
3852 record.target_ref.clone(),
3853 target,
3854 source,
3855 )];
3856 let keys = crate::transport::key_directory(&self.stores);
3857 let ctx = gunnar_policy::PushContext::new(
3858 &policy_principal,
3859 &repo_id,
3860 &updates,
3861 &introduced,
3862 git,
3863 &keys,
3864 );
3865 match gunnar_policy::Gate::check(&gate, &ctx) {
3866 Ok(gunnar_policy::Verdict::Allow) => {}
3867 Ok(verdict) => {
3868 return Err(ControlError::FailedPrecondition(format!(
3869 "branch protection refuses this merge: {}",
3870 verdict.reason().unwrap_or_default()
3871 )));
3872 }
3873 Err(err) => {
3876 return Err(ControlError::FailedPrecondition(format!(
3877 "branch protection could not decide this merge ({err}); refusing rather \
3878 than guessing"
3879 )));
3880 }
3881 }
3882 }
3883
3884 if source != target {
3885 match gunnar_policy::ancestry::is_descendant_of(
3886 git,
3887 &source,
3888 &target,
3889 gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET,
3890 ) {
3891 Ok(true) => {}
3892 Ok(false) => {
3893 return Err(ControlError::FailedPrecondition(format!(
3894 "not a fast-forward: {} does not descend from {}. gunnar has no merge \
3895 engine yet — rebase or merge locally, push the result, and merge again",
3896 record.source_ref, record.target_ref
3897 )));
3898 }
3899 Err(err) => {
3902 return Err(ControlError::FailedPrecondition(format!(
3903 "the ancestry walk could not be completed ({err}); refusing rather than \
3904 guessing"
3905 )));
3906 }
3907 }
3908 git.update_ref(
3909 &record.target_ref,
3910 Some(target.as_bytes()),
3911 Some(source.as_bytes()),
3912 )
3913 .map_err(|err| {
3914 ControlError::FailedPrecondition(format!(
3915 "the target ref moved while merging ({err}); nothing was written — retry \
3916 against the new tip"
3917 ))
3918 })?;
3919 }
3920 let merged = match pulls
3924 .mark_merged(&key, number, &source.to_string())
3925 .map_err(internal)?
3926 {
3927 gunnar_core::PullTransition::Done(record) => record,
3928 other => {
3931 return Err(ControlError::Internal(format!(
3932 "the ref moved to {source} but the record could not be marked merged \
3933 ({other:?}); the repository is correct, the record is not"
3934 )));
3935 }
3936 };
3937 Ok((
3938 pull_out(path, merged),
3939 MergedRef {
3940 ref_name: record.target_ref.clone(),
3941 old_oid: target.to_string(),
3942 new_oid: source.to_string(),
3943 },
3944 ))
3945 }
3946
3947 fn read_object(
3948 &self,
3949 path: &RepoPath,
3950 oid: &str,
3951 max_bytes: u64,
3952 ) -> ControlResult<ReadObjectResponse> {
3953 let (_, store) = self.open(path)?;
3954 let git = store.serve();
3955 let id = gix_hash::ObjectId::from_hex(oid.as_bytes())
3956 .map_err(|err| ControlError::InvalidArgument(format!("{oid}: {err}")))?;
3957
3958 if let Some((_, object_bytes)) = git.header(id.as_bytes()).map_err(store_error)? {
3966 if self.max_object_bytes != 0 && object_bytes > self.max_object_bytes {
3971 return Err(ControlError::ResourceExhausted(format!(
3972 "object {oid} is {object_bytes} bytes, over this server's {}-byte ceiling \
3973 for reading an object. Reading any part of it costs the whole object in \
3974 memory, because a prefix of a delta-encoded object needs its entire base \
3975 chain, so no max_bytes makes this request cheaper.",
3976 self.max_object_bytes
3977 )));
3978 }
3979 let response_bytes = if max_bytes == 0 {
3982 object_bytes
3983 } else {
3984 object_bytes.min(max_bytes)
3985 };
3986 if response_bytes > MAX_RESPONSE_BYTES {
3987 return Err(ControlError::ResourceExhausted(format!(
3988 "object {oid} is {object_bytes} bytes and the response would be \
3989 {response_bytes}, over the {MAX_RESPONSE_BYTES}-byte limit a gRPC client \
3990 can receive by default. Ask for a prefix with max_bytes."
3991 )));
3992 }
3993 }
3994
3995 let Some((kind, mut buf)) = git.read(id.as_bytes()).map_err(store_error)? else {
4005 return Err(ControlError::NotFound(format!("object {oid} in {path}")));
4006 };
4007 let size = buf.len() as u64;
4008
4009 let truncated = max_bytes != 0 && size > max_bytes;
4015 if truncated {
4016 buf.truncate(max_bytes as usize);
4017 }
4018 Ok(ReadObjectResponse {
4019 oid: id.to_string(),
4020 kind: object_kind_out(kind)? as i32,
4021 size,
4022 data: buf,
4023 truncated,
4024 })
4025 }
4026
4027 fn read_lfs_object(
4059 &self,
4060 path: &RepoPath,
4061 oid: &str,
4062 max_bytes: u64,
4063 range: Option<(u64, u64)>,
4064 ) -> ControlResult<ReadLfsObjectResponse> {
4065 let (_, store) = self.open(path)?;
4066 let id = gunnar_lfs::Oid::parse(oid)
4067 .map_err(|err| ControlError::InvalidArgument(format!("{oid}: {err}")))?;
4068 use gunnar_lfs::LfsStore as _;
4071
4072 let lfs = gunnar_lfs::FsLfsStore::open_or_init(store.git_dir()).map_err(|err| {
4073 ControlError::Internal(format!("opening the repository's LFS store: {err}"))
4074 })?;
4075
4076 let ceiling = match max_bytes {
4077 0 => MAX_RESPONSE_BYTES,
4078 n => n.min(MAX_RESPONSE_BYTES),
4079 };
4080 let size = lfs
4081 .size_of(&id)
4082 .map_err(|err| ControlError::Internal(format!("reading the LFS store: {err}")))?
4083 .ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
4084
4085 if let Some((offset, length)) = range {
4092 if offset > size || (offset == size && size > 0) {
4093 return Err(ControlError::InvalidArgument(format!(
4094 "offset {offset} is past the end of LFS object {oid}, which is {size} bytes"
4095 )));
4096 }
4097 if length > ceiling {
4098 return Err(ControlError::ResourceExhausted(format!(
4099 "a range of {length} bytes of LFS object {oid} is over the {ceiling}-byte ceiling \
4100 for this request; ask for {ceiling} bytes or fewer at a time"
4101 )));
4102 }
4103 let want = if length == 0 { ceiling } else { length }.min(size - offset);
4104 let object = lfs
4105 .open_at(&id, offset)
4106 .map_err(|err| ControlError::Internal(format!("opening LFS object {oid}: {err}")))?
4107 .ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
4108 let mut data = Vec::with_capacity(want as usize);
4109 let mut reader = std::io::Read::take(object.reader, want);
4110 std::io::Read::read_to_end(&mut reader, &mut data)
4111 .map_err(|err| ControlError::Internal(format!("reading LFS object {oid}: {err}")))?;
4112 if (data.len() as u64) < want {
4113 return Err(ControlError::Internal(format!(
4114 "LFS object {oid} ended {} bytes into a {want}-byte range at {offset}",
4115 data.len()
4116 )));
4117 }
4118 return Ok(ReadLfsObjectResponse { oid: oid.to_owned(), size: object.size, data, offset });
4119 }
4120
4121 if size > ceiling {
4122 return Err(ControlError::ResourceExhausted(format!(
4123 "LFS object {oid} is {size} bytes, over the {ceiling}-byte ceiling for this \
4124 request. An LFS object is served whole or not at all — a prefix of one is not a \
4125 smaller file — so no max_bytes makes this request smaller. `git lfs` streams it, \
4126 and a ranged read (offset/length) reads it a range at a time."
4127 )));
4128 }
4129
4130 let object = lfs
4131 .open(&id)
4132 .map_err(|err| ControlError::Internal(format!("opening LFS object {oid}: {err}")))?
4133 .ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
4134 if object.size > ceiling {
4139 return Err(ControlError::ResourceExhausted(format!(
4140 "LFS object {oid} is {} bytes, over the {ceiling}-byte ceiling for this request",
4141 object.size
4142 )));
4143 }
4144 let mut data = Vec::with_capacity(object.size as usize);
4145 let mut reader = std::io::Read::take(object.reader, object.size);
4146 std::io::Read::read_to_end(&mut reader, &mut data)
4147 .map_err(|err| ControlError::Internal(format!("reading LFS object {oid}: {err}")))?;
4148
4149 Ok(ReadLfsObjectResponse {
4150 oid: oid.to_owned(),
4151 size: data.len() as u64,
4152 data,
4153 offset: 0,
4154 })
4155 }
4156
4157 fn diff(
4166 &self,
4167 path: &RepoPath,
4168 from: &str,
4169 to: &str,
4170 path_prefixes: &[String],
4171 ) -> ControlResult<Vec<FileChange>> {
4172 let (_repo, store) = self.open(path)?;
4175 let git = store.serve();
4176 let hash = store.hash_kind();
4177
4178 let resolve = |spec: &str| -> ControlResult<Option<gix_hash::ObjectId>> {
4182 if spec.is_empty() {
4183 return Ok(None);
4184 }
4185 Ok(Some(self.tree_of(git, hash, path, spec)?))
4186 };
4187 let old = resolve(from)?;
4188 let new = resolve(to)?;
4189
4190 crate::diff::compare(git, hash, old, new, path_prefixes).map_err(|err| match err {
4191 crate::diff::DiffError::Missing(oid) => {
4194 ControlError::NotFound(format!("object {oid} in {path}"))
4195 }
4196 crate::diff::DiffError::WrongKind { .. } => {
4197 ControlError::InvalidArgument(err.to_string())
4198 }
4199 crate::diff::DiffError::Corrupt { .. } | crate::diff::DiffError::Backend(_) => {
4200 ControlError::Internal(err.to_string())
4201 }
4202 })
4203 }
4204
4205 fn add_principal(
4208 &self,
4209 id: &str,
4210 name: &str,
4211 kind: gunnar_proto::PrincipalKind,
4212 ) -> ControlResult<Principal> {
4213 let principal = gunnar_core::Principal::new(id, name, principal_kind_in(kind))
4224 .map_err(internal)?;
4225 let created = self
4226 .stores
4227 .auth()
4228 .add_principal(principal)
4229 .map(|p| principal_out(&p))
4230 .map_err(internal)?;
4231 if let Err(err) = self.stores.catalog().create_namespace(id) {
4245 let _ = self.stores.auth().remove_principal(id);
4246 return Err(ControlError::AlreadyExists(format!(
4247 "cannot create the account namespace \"{id}\" for this user: {err}. A user's \
4248 account is the namespace named after their id and cannot be somebody else's; \
4249 the principal was not created."
4250 )));
4251 }
4252 self.stores.mirror_identity();
4256 Ok(created)
4257 }
4258
4259 fn enroll_employee(
4260 &self,
4261 account: &str,
4262 ask: &gunnar_grpc::EmployeeAsk,
4263 ) -> ControlResult<gunnar_grpc::EmployeeAnswer> {
4264 if self.stores.catalog().namespace(account).map_err(internal)?.is_none() {
4265 return Err(ControlError::NotFound(format!("account {account}")));
4266 }
4267 if self.stores.auth().is_clan(account).map_err(internal)? {
4270 return Err(ControlError::FailedPrecondition(format!(
4271 "{account} is a clan, not an account: employees are enrolled by an account's admin, \
4272 and a clan's members are accounts that join it"
4273 )));
4274 }
4275 let principal = if ask.seats == 0 {
4276 if self.stores.catalog().namespace(&ask.id).map_err(internal)?.is_some() {
4280 return Err(ControlError::AlreadyExists(format!(
4281 "namespace {} exists, so an employee may not be called that",
4282 ask.id
4283 )));
4284 }
4285 Some(
4286 gunnar_core::Principal::new(&ask.id, &ask.name, gunnar_core::PrincipalKind::User)
4287 .map_err(internal)?,
4288 )
4289 } else {
4290 None
4291 };
4292 let room = gunnar_core::RoomAsk {
4293 room_bytes: ask.room_bytes,
4294 seats: ask.seats,
4295 from_seat: ask.from_seat,
4296 cap: ask.cap,
4297 stored: ask.stored,
4298 order: ask.order.clone(),
4299 dry_run: ask.dry_run,
4300 };
4301 let answer = match self.stores.auth().enroll_with_room(account, principal, &room) {
4302 Ok(answer) => answer,
4303 Err(gunnar_core::Error::RoomShort { need, unused, short, .. }) => {
4306 let (seats, seats_open) =
4307 self.stores.auth().seats_of(account).map_err(internal)?;
4308 return Ok(gunnar_grpc::EmployeeAnswer {
4309 principal: None,
4310 room_bytes: ask.room_bytes,
4311 seats,
4312 seats_open,
4313 need_bytes: need,
4314 unused_bytes: Some(unused),
4315 shortfall_bytes: short,
4316 unchanged: true,
4317 });
4318 }
4319 Err(gunnar_core::Error::PrincipalExists(id)) => {
4320 return Err(ControlError::AlreadyExists(format!(
4321 "principal {id} already exists, and an existing principal can never become an \
4322 employee"
4323 )))
4324 }
4325 Err(err @ gunnar_core::Error::NoOpenSeat(_)) => {
4326 return Err(ControlError::FailedPrecondition(err.to_string()))
4327 }
4328 Err(err @ gunnar_core::Error::NameExists { .. }) => {
4329 return Err(ControlError::AlreadyExists(err.to_string()))
4330 }
4331 Err(err @ gunnar_core::Error::InvalidInput { .. }) => {
4332 return Err(ControlError::InvalidArgument(err.to_string()))
4333 }
4334 Err(other) => return Err(internal(other)),
4335 };
4336 if !answer.unchanged {
4337 self.stores.mirror_identity();
4338 }
4339 Ok(gunnar_grpc::EmployeeAnswer {
4340 principal: answer.principal.as_ref().map(principal_out),
4341 room_bytes: answer.room_bytes,
4342 seats: answer.seats,
4343 seats_open: answer.seats_open,
4344 need_bytes: answer.need_bytes,
4345 unused_bytes: answer.unused_bytes,
4346 shortfall_bytes: answer.shortfall_bytes,
4347 unchanged: answer.unchanged,
4348 })
4349 }
4350
4351 fn seats(&self, account: &str) -> ControlResult<(u32, u32)> {
4352 self.stores.auth().seats_of(account).map_err(internal)
4353 }
4354
4355 fn clans(&self) -> Option<&dyn gunnar_grpc::clans::ClanPlane> {
4356 Some(self)
4357 }
4358
4359 fn has_employees(&self, account: &str) -> ControlResult<bool> {
4360 self.stores.auth().is_enterprise(account).map_err(internal)
4361 }
4362
4363 fn employees(&self, account: &str) -> ControlResult<Vec<Principal>> {
4364 Ok(self
4365 .stores
4366 .auth()
4367 .employees_of(account)
4368 .map_err(internal)?
4369 .iter()
4370 .map(principal_out)
4371 .collect())
4372 }
4373
4374 fn principal_by_name(&self, name: &str) -> ControlResult<Option<Principal>> {
4375 Ok(self
4376 .stores
4377 .auth()
4378 .principal_by_name(name)
4379 .map_err(internal)?
4380 .as_ref()
4381 .map(principal_out))
4382 }
4383
4384 fn operator_bootstrap_invite(
4396 &self,
4397 operator: &str,
4398 secret_sha256: &str,
4399 now_unix_ms: i64,
4400 ttl_ms: i64,
4401 ) -> ControlResult<gunnar_grpc::BootstrapInvite> {
4402 use gunnar_grpc::seam::OPERATOR_INVITE_ID;
4403 let hash = gunnar_grpc::seam::bootstrap_hash(secret_sha256)?;
4404 let ttl_ms = gunnar_grpc::seam::bootstrap_ttl(ttl_ms)?;
4405 let auth = self.stores.auth();
4406 if !auth.web_credentials(Some(operator)).map_err(internal)?.is_empty() {
4407 if auth.revoke_invite(OPERATOR_INVITE_ID).map_err(internal)?.is_some() {
4408 self.stores.mirror_identity();
4409 }
4410 return Ok(gunnar_grpc::BootstrapInvite::Spent);
4411 }
4412 let renewed = auth
4413 .invites()
4414 .map_err(internal)?
4415 .iter()
4416 .any(|row| row.id == OPERATOR_INVITE_ID && row.secret_hash == hash && row.principal_id == operator);
4417 auth.revoke_invite(OPERATOR_INVITE_ID).map_err(internal)?;
4418 auth.create_invite(gunnar_core::Invite {
4419 id: OPERATOR_INVITE_ID.to_owned(),
4420 principal_id: operator.to_owned(),
4421 secret_hash: hash,
4422 created_by: "installer".to_owned(),
4423 created: now_unix_ms,
4424 expires: now_unix_ms + ttl_ms,
4425 note: "the operator's first passkey: the installer's first invite".to_owned(),
4426 server_level: None,
4427 })
4428 .map_err(internal)?;
4429 self.stores.mirror_identity();
4430 if !renewed {
4431 tracing::warn!(
4432 target: "gunnar::audit",
4433 event = "operator.bootstrap_invite",
4434 operator,
4435 ttl_days = ttl_ms / 86_400_000,
4436 "the installer's first invite (its hash) is recorded for the operator: his first passkey registers on his account, by gunnar's word"
4437 );
4438 }
4439 Ok(if renewed { gunnar_grpc::BootstrapInvite::Renewed } else { gunnar_grpc::BootstrapInvite::Recorded })
4440 }
4441
4442 fn claim_operator(&self, asked: &str) -> ControlResult<gunnar_grpc::OperatorClaim> {
4443 #[cfg(feature = "wg-client")]
4444 let fed = crate::fed_principals::fed_operator(self.stores.root());
4445 #[cfg(not(feature = "wg-client"))]
4446 let fed: Option<String> = None;
4447 self.claim_operator_named(fed.as_deref(), asked)
4448 }
4449
4450 fn rename_principal(&self, principal_id: &str, name: &str) -> ControlResult<Principal> {
4451 let renamed = self
4455 .stores
4456 .auth()
4457 .rename_principal(principal_id, name)
4458 .map(|p| principal_out(&p))
4459 .map_err(internal)?;
4460 self.stores.mirror_identity();
4461 Ok(renamed)
4462 }
4463
4464 fn list_principals(&self) -> ControlResult<Vec<Principal>> {
4465 Ok(self
4466 .stores
4467 .auth()
4468 .principals()
4469 .map_err(internal)?
4470 .iter()
4471 .map(principal_out)
4472 .collect())
4473 }
4474
4475 fn principal_activity(
4476 &self,
4477 ) -> ControlResult<std::collections::BTreeMap<String, gunnar_grpc::PrincipalSeen>> {
4478 Ok(self
4479 .stores
4480 .activity()
4481 .all()
4482 .into_iter()
4483 .map(|(id, seen)| {
4484 (
4485 id,
4486 gunnar_grpc::PrincipalSeen {
4487 last_sign_in_unix_ms: seen.signed_in,
4488 last_operation_unix_ms: seen.operated,
4489 },
4490 )
4491 })
4492 .collect())
4493 }
4494
4495 fn holds_server_grant(&self, principal_id: &str) -> ControlResult<bool> {
4496 self.stores.auth().holds_server_grant(principal_id).map_err(internal)
4497 }
4498
4499 fn add_credential(
4500 &self,
4501 principal_id: &str,
4502 credential_id: &[u8],
4503 public_key: &[u8],
4504 label: &str,
4505 sign_count: u32,
4506 ) -> ControlResult<gunnar_proto::WebCredential> {
4507 let stored = self
4508 .stores
4509 .auth()
4510 .add_web_credential(gunnar_core::WebCredential {
4511 id: gunnar_grpc::webauthn::b64url(credential_id),
4512 principal_id: principal_id.to_owned(),
4513 label: label.to_owned(),
4514 public_key_hex: hex::encode(public_key),
4515 sign_count,
4516 created: gunnar_core::now_ms(),
4517 last_used: None,
4518 sessions: Vec::new(),
4519 })
4520 .map_err(internal)?;
4521 web_credential_out(&stored)
4522 }
4523
4524 fn credentials(&self, principal_id: Option<&str>) -> ControlResult<Vec<gunnar_proto::WebCredential>> {
4525 self.stores
4526 .auth()
4527 .web_credentials(principal_id)
4528 .map_err(internal)?
4529 .iter()
4530 .map(web_credential_out)
4531 .collect()
4532 }
4533
4534 fn credential(&self, credential_id: &[u8]) -> ControlResult<Option<gunnar_grpc::StoredCredential>> {
4535 let Some(row) = self
4536 .stores
4537 .auth()
4538 .web_credential(&gunnar_grpc::webauthn::b64url(credential_id))
4539 .map_err(internal)?
4540 else {
4541 return Ok(None);
4542 };
4543 let public_key = hex::decode(&row.public_key_hex).map_err(|err| {
4544 ControlError::Internal(format!("web credential {} holds a key that is not hex: {err}", row.id))
4545 })?;
4546 Ok(Some(gunnar_grpc::StoredCredential {
4547 credential: web_credential_out(&row)?,
4548 public_key,
4549 }))
4550 }
4551
4552 fn revoke_credential(&self, credential_id: &[u8]) -> ControlResult<Option<gunnar_proto::WebCredential>> {
4553 let removed = self
4554 .stores
4555 .auth()
4556 .revoke_web_credential(&gunnar_grpc::webauthn::b64url(credential_id))
4557 .map_err(internal)?;
4558 self.stores.mirror_identity();
4560 removed.as_ref().map(web_credential_out).transpose()
4561 }
4562
4563 fn rename_credential(
4564 &self,
4565 credential_id: &[u8],
4566 label: &str,
4567 ) -> ControlResult<Option<gunnar_proto::WebCredential>> {
4568 self.stores
4569 .auth()
4570 .rename_web_credential(&gunnar_grpc::webauthn::b64url(credential_id), label)
4571 .map_err(internal)?
4572 .as_ref()
4573 .map(web_credential_out)
4574 .transpose()
4575 }
4576
4577 fn create_invite(&self, invite: gunnar_grpc::NewInvite) -> ControlResult<gunnar_proto::Invite> {
4578 if invite.server_level == gunnar_proto::GrantLevel::Admin {
4580 return Err(ControlError::InvalidArgument(gunnar_grpc::service::access::NO_ADMIN_BY_INVITE.to_owned()));
4581 }
4582 let stored = self
4583 .stores
4584 .auth()
4585 .create_invite(gunnar_core::Invite {
4586 id: invite.id,
4587 principal_id: invite.principal_id,
4588 secret_hash: invite.secret_hash,
4589 created_by: invite.created_by,
4590 created: invite.created_at_unix_ms,
4591 expires: invite.expires_at_unix_ms,
4592 note: invite.note,
4593 server_level: level_in_opt(invite.server_level),
4594 })
4595 .map_err(internal)?;
4596 Ok(invite_out(&stored))
4597 }
4598
4599 fn invites(&self) -> ControlResult<Vec<gunnar_proto::Invite>> {
4600 Ok(self
4601 .stores
4602 .auth()
4603 .invites()
4604 .map_err(internal)?
4605 .iter()
4606 .map(invite_out)
4607 .collect())
4608 }
4609
4610 fn revoke_invite(&self, id: &str) -> ControlResult<Option<gunnar_proto::Invite>> {
4611 Ok(self
4612 .stores
4613 .auth()
4614 .revoke_invite(id)
4615 .map_err(internal)?
4616 .as_ref()
4617 .map(invite_out))
4618 }
4619
4620 fn check_invite(&self, secret_hash: &str, now_unix_ms: i64) -> ControlResult<gunnar_grpc::InviteCheck> {
4621 Ok(match self.stores.auth().check_invite(secret_hash, now_unix_ms).map_err(internal)? {
4622 gunnar_core::InviteStanding::Valid(invite) => gunnar_grpc::InviteCheck::Valid(invite_out(&invite)),
4623 gunnar_core::InviteStanding::Used(_) => gunnar_grpc::InviteCheck::Used,
4624 gunnar_core::InviteStanding::Expired(_) => gunnar_grpc::InviteCheck::Expired,
4625 gunnar_core::InviteStanding::Unknown => gunnar_grpc::InviteCheck::Unknown,
4626 })
4627 }
4628
4629 fn employer_of(&self, principal_id: &str) -> ControlResult<Option<String>> {
4630 self.stores.auth().employer_of(principal_id).map_err(internal)
4631 }
4632
4633 fn sso(&self) -> Option<&dyn gunnar_grpc::sso::plane::SsoPlane> {
4634 Some(self)
4635 }
4636
4637 fn web_login(&self, commit: gunnar_grpc::WebLoginCommit) -> ControlResult<gunnar_grpc::WebLoginOutcome> {
4638 let (principal, credential, key) = self
4639 .stores
4640 .auth()
4641 .web_login(
4642 &gunnar_grpc::webauthn::b64url(&commit.credential_id),
4643 commit.sign_count,
4644 commit.now_unix_ms,
4645 &commit.session_openssh,
4646 commit.expires_at_unix_ms,
4647 )
4648 .map_err(internal)?;
4649 self.stores.mirror_identity();
4650 self.stores.activity().signed_in(&principal.id, commit.now_unix_ms);
4652 Ok(gunnar_grpc::WebLoginOutcome {
4653 principal: principal_out(&principal),
4654 credential: web_credential_out(&credential)?,
4655 fingerprint: key.fingerprint_sha256,
4656 })
4657 }
4658
4659 fn web_register(&self, registration: gunnar_grpc::WebRegistration) -> ControlResult<gunnar_grpc::WebRegistered> {
4660 let (credential, principal, level) = self
4661 .stores
4662 .auth()
4663 .register_web_credential(
4664 registration.invite_secret_hash.as_deref(),
4665 registration.principal_id.as_deref(),
4666 gunnar_core::WebCredential {
4667 id: gunnar_grpc::webauthn::b64url(®istration.credential_id),
4668 principal_id: String::new(),
4669 label: registration.label,
4670 public_key_hex: hex::encode(®istration.public_key),
4671 sign_count: registration.sign_count,
4672 created: registration.now_unix_ms,
4673 last_used: None,
4674 sessions: Vec::new(),
4675 },
4676 registration.now_unix_ms,
4677 )
4678 .map_err(internal)?;
4679 Ok(gunnar_grpc::WebRegistered {
4680 credential: web_credential_out(&credential)?,
4681 principal: principal_out(&principal),
4682 server_level: level.map(level_out).unwrap_or(GrantLevel::Unspecified),
4683 })
4684 }
4685
4686 fn approve_machine_enrolment(
4687 &self,
4688 approval: gunnar_grpc::MachineApproval,
4689 ) -> ControlResult<gunnar_grpc::MachineApproved> {
4690 let step_up = approval
4691 .step_up
4692 .as_ref()
4693 .map(|(id, count)| (gunnar_grpc::webauthn::b64url(id), *count));
4694 let (principal, credential, created) = self
4695 .stores
4696 .auth()
4697 .approve_machine_credential(
4698 &approval.machine,
4699 level_in(approval.server_level)?,
4700 gunnar_core::WebCredential {
4701 id: gunnar_grpc::webauthn::b64url(&approval.credential_id),
4702 principal_id: String::new(),
4703 label: approval.label,
4704 public_key_hex: hex::encode(&approval.public_key),
4705 sign_count: approval.sign_count,
4706 created: approval.now_unix_ms,
4707 last_used: None,
4708 sessions: Vec::new(),
4709 },
4710 step_up.as_ref().map(|(id, count)| (id.as_str(), *count)),
4711 approval.now_unix_ms,
4712 )
4713 .map_err(internal)?;
4714 self.stores.mirror_identity();
4715 Ok(gunnar_grpc::MachineApproved {
4716 principal: principal_out(&principal),
4717 credential: web_credential_out(&credential)?,
4718 created,
4719 })
4720 }
4721
4722 fn remove_principal(&self, principal_id: &str) -> ControlResult<(u32, u32)> {
4723 let before_keys = self.stores.auth().keys().map_err(internal)?.len();
4724 let before_grants = self.stores.auth().grants(None).map_err(internal)?.len();
4725 self.stores
4726 .auth()
4727 .remove_principal(principal_id)
4728 .map_err(internal)?;
4729 let after_keys = self.stores.auth().keys().map_err(internal)?.len();
4730 let after_grants = self.stores.auth().grants(None).map_err(internal)?.len();
4731 self.stores.mirror_identity();
4732 Ok((
4733 before_keys.saturating_sub(after_keys) as u32,
4734 before_grants.saturating_sub(after_grants) as u32,
4735 ))
4736 }
4737
4738 fn add_key(
4739 &self,
4740 principal_id: &str,
4741 openssh_public_key: &str,
4742 _comment: &str,
4743 expires_at_unix_ms: Option<i64>,
4744 caller: &str,
4745 attestation: Option<gunnar_proto::SignedKeyAttestation>,
4746 ) -> ControlResult<PublicKey> {
4747 let adding = match gunnar_core::auth::pq_recipient_fingerprint(openssh_public_key) {
4755 Some(fingerprint) => fingerprint,
4756 None => gunnar_core::Fingerprint::of_key(
4757 &gunnar_core::auth::parse_public_key(openssh_public_key).map_err(internal)?,
4758 ),
4759 };
4760 let holders = self
4761 .stores
4762 .auth()
4763 .attesting_keys_of(principal_id)
4764 .map_err(internal)?;
4765 if holders.iter().any(|k| k.fingerprint == adding.as_str()) {
4769 return Err(ControlError::AlreadyExists(format!("principal {principal_id} already holds key {}", adding.as_str())));
4770 }
4771 let requirement = gunnar_core::key_attestation::Requirement::of(principal_id, caller, holders);
4772 let attestation = attestation.map(from_proto_attestation).transpose()?;
4773 gunnar_core::key_attestation::check(
4774 &requirement,
4775 principal_id,
4776 adding.as_str(),
4777 attestation.as_ref(),
4778 )
4779 .map_err(|problem| match problem {
4780 gunnar_core::AttestationProblem::Missing { .. } => {
4784 ControlError::FailedPrecondition(problem.to_string())
4785 }
4786 _ => ControlError::InvalidArgument(problem.to_string()),
4788 })?;
4789 let registered = self
4790 .stores
4791 .auth()
4792 .register_key(
4793 principal_id,
4794 openssh_public_key,
4795 expires_at_unix_ms.map(|ms| ms as UnixMillis),
4796 )
4797 .map(|key| key_out(&key))
4798 .map_err(internal)?;
4799 self.stores.mirror_identity();
4800 Ok(registered)
4801 }
4802
4803 fn revoke_key(&self, fingerprint: &str) -> ControlResult<()> {
4804 let parsed = gunnar_core::Fingerprint::parse(fingerprint)
4805 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
4806 let revoked = self.stores.auth().revoke_key(&parsed).map_err(internal)?;
4807 self.stores.mirror_identity();
4808 match revoked {
4809 Some(_) => Ok(()),
4810 None => Err(ControlError::NotFound(format!("key {fingerprint}"))),
4811 }
4812 }
4813
4814 fn list_keys(&self, principal_id: Option<&str>) -> ControlResult<Vec<PublicKey>> {
4815 let keys = match principal_id {
4816 Some(id) => self.stores.auth().keys_of(id).map_err(internal)?,
4817 None => self.stores.auth().keys().map_err(internal)?,
4818 };
4819 Ok(keys.iter().map(key_out).collect())
4820 }
4821
4822 fn grant(
4823 &self,
4824 principal_id: &str,
4825 namespace: &str,
4826 repo: &str,
4827 level: GrantLevel,
4828 ) -> ControlResult<gunnar_proto::Grant> {
4829 let scope = scope_in(namespace, repo)?;
4830 let grant = Grant {
4831 principal: principal_id.to_owned(),
4832 scope,
4833 level: level_in(level)?,
4834 };
4835 self.stores
4836 .auth()
4837 .grant(grant)
4838 .map(|g| grant_out(&g))
4839 .map_err(internal)
4840 }
4841
4842 fn revoke(&self, principal_id: &str, namespace: &str, repo: &str) -> ControlResult<()> {
4843 let scope = scope_in(namespace, repo)?;
4844 match self
4845 .stores
4846 .auth()
4847 .revoke_grant(principal_id, &scope)
4848 .map_err(internal)?
4849 {
4850 true => Ok(()),
4851 false => Err(ControlError::NotFound(format!(
4852 "grant for {principal_id} on {scope}"
4853 ))),
4854 }
4855 }
4856
4857 fn list_grants(
4858 &self,
4859 principal_id: Option<&str>,
4860 namespace: Option<&str>,
4861 repo: Option<&str>,
4862 ) -> ControlResult<Vec<gunnar_proto::Grant>> {
4863 Ok(self
4864 .stores
4865 .auth()
4866 .grants(principal_id)
4867 .map_err(internal)?
4868 .iter()
4869 .map(grant_out)
4870 .filter(|g| namespace.is_none_or(|want| want == g.namespace))
4871 .filter(|g| repo.is_none_or(|want| want == g.repo))
4872 .collect())
4873 }
4874
4875 fn add_group_member(
4878 &self,
4879 group_id: &str,
4880 member_id: &str,
4881 signed: gunnar_proto::SignedMembershipEvent,
4882 ) -> ControlResult<u64> {
4883 let signed = self.verify_membership_event(
4884 signed,
4885 group_id,
4886 member_id,
4887 gunnar_core::MembershipAction::Add,
4888 )?;
4889 self.stores
4893 .auth()
4894 .add_to_group(group_id, member_id)
4895 .map_err(internal)?;
4896 self.append_membership_event(&signed)
4897 }
4898
4899 fn remove_group_member(
4900 &self,
4901 group_id: &str,
4902 member_id: &str,
4903 signed: gunnar_proto::SignedMembershipEvent,
4904 ) -> ControlResult<(bool, u64)> {
4905 let signed = self.verify_membership_event(
4906 signed,
4907 group_id,
4908 member_id,
4909 gunnar_core::MembershipAction::Remove,
4910 )?;
4911 let was_member = self
4912 .stores
4913 .auth()
4914 .remove_from_group(group_id, member_id)
4915 .map_err(internal)?;
4916 let index = self.append_membership_event(&signed)?;
4920 Ok((was_member, index))
4921 }
4922
4923 fn membership_events(
4924 &self,
4925 group_id: Option<&str>,
4926 ) -> ControlResult<Vec<gunnar_proto::SignedMembershipEvent>> {
4927 Ok(self
4928 .membership_log_events()?
4929 .into_iter()
4930 .filter(|signed| group_id.is_none_or(|g| signed.event.group == g))
4931 .map(to_proto_membership)
4932 .collect())
4933 }
4934
4935 fn list_group_members(&self, group_id: &str) -> ControlResult<Vec<Principal>> {
4936 let auth = self.stores.auth();
4939 match auth.principal(group_id).map_err(internal)? {
4940 Some(p) if p.kind == PrincipalKind::Group => {}
4941 Some(_) => {
4942 return Err(ControlError::InvalidArgument(format!(
4943 "principal \"{group_id}\" is not a group"
4944 )))
4945 }
4946 None => return Err(ControlError::NotFound(format!("group {group_id}"))),
4947 }
4948 let mut members = Vec::new();
4951 for id in auth.members_of(group_id).map_err(internal)? {
4952 if let Some(p) = auth.principal(&id).map_err(internal)? {
4953 members.push(principal_out(&p));
4954 }
4955 }
4956 Ok(members)
4957 }
4958
4959 fn get_protection(
4962 &self,
4963 path: &RepoPath,
4964 ref_pattern: Option<&str>,
4965 ) -> ControlResult<Vec<BranchProtection>> {
4966 let (repo, _) = self.open(path)?;
4967 let config = self.policy_of(&repo)?;
4968 Ok(config
4969 .protection
4970 .iter()
4971 .filter(|rule| ref_pattern.is_none_or(|want| rule.pattern.as_str() == want))
4972 .map(protection_out)
4973 .collect())
4974 }
4975
4976 fn set_protection(
4996 &self,
4997 path: &RepoPath,
4998 protection: BranchProtection,
4999 ) -> ControlResult<BranchProtection> {
5000 let (repo, _) = self.open(path)?;
5001 if protection.max_object_bytes != 0 || protection.max_objects_per_push != 0 {
5002 return Err(ControlError::InvalidArgument(
5003 "max_object_bytes and max_objects_per_push are per-rule in the protocol and \
5004 repository-wide in this server's policy engine, so a value here would apply to \
5005 references this rule does not name. Leave them 0 and set the repository's \
5006 limits in its policy document."
5007 .to_owned(),
5008 ));
5009 }
5010 let pattern = gunnar_policy::RefPattern::new(protection.ref_pattern.clone())
5011 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
5012
5013 let mut config = self.policy_of(&repo)?;
5014 let mut rule = config
5017 .protection
5018 .iter()
5019 .find(|rule| rule.pattern == pattern)
5020 .cloned()
5021 .unwrap_or_else(|| {
5022 gunnar_policy::ProtectionRule::protect(pattern.as_str())
5023 .expect("the pattern was compiled one line above")
5024 });
5025 rule.pattern = pattern;
5026 rule.allow_force_push = protection.allow_force_push;
5027 rule.allow_delete = protection.allow_delete;
5028 rule.require_signature = protection.require_signature;
5029 rule.allowed_principals = if protection.allowed_principal_ids.is_empty() {
5033 None
5034 } else {
5035 Some(protection.allowed_principal_ids.clone())
5036 };
5037 rule.forbid_self_approval = protection.forbid_self_approval;
5038 rule.required_approvals = protection.required_approvals;
5039
5040 config
5041 .protection
5042 .retain(|other| other.pattern != rule.pattern);
5043 config.protection.push(rule.clone());
5044 self.store_policy(&repo, &config)?;
5045 Ok(protection_out(&rule))
5046 }
5047
5048 fn delete_protection(&self, path: &RepoPath, ref_pattern: &str) -> ControlResult<()> {
5049 let (repo, _) = self.open(path)?;
5050 let mut config = self.policy_of(&repo)?;
5051 let before = config.protection.len();
5052 config
5053 .protection
5054 .retain(|rule| rule.pattern.as_str() != ref_pattern);
5055 if config.protection.len() == before {
5056 return Err(ControlError::NotFound(format!(
5057 "no protection rule for {ref_pattern} in {path}"
5058 )));
5059 }
5060 self.store_policy(&repo, &config)
5061 }
5062
5063 fn get_ref_name_policy(&self, path: &RepoPath) -> ControlResult<RefNamePolicy> {
5064 let (repo, _) = self.open(path)?;
5065 let config = self.policy_of(&repo)?;
5066 Ok(ref_name_policy_out(&config.ref_names))
5067 }
5068
5069 fn set_ref_name_policy(
5070 &self,
5071 path: &RepoPath,
5072 policy: RefNamePolicy,
5073 ) -> ControlResult<RefNamePolicy> {
5074 let (repo, _) = self.open(path)?;
5075 let compiled = gunnar_policy::RefNamePolicy::new(
5076 policy.allow_patterns.clone(),
5077 policy.deny_patterns.clone(),
5078 )
5079 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
5080 let mut config = self.policy_of(&repo)?;
5081 config.ref_names = compiled.clone();
5082 self.store_policy(&repo, &config)?;
5083 Ok(ref_name_policy_out(&compiled))
5084 }
5085
5086 fn forge_append(&self, event: gunnar_proto::SignedForgeEvent) -> ControlResult<u64> {
5094 let signed = from_proto_event(event)?;
5095
5096 let key = self
5103 .stores
5104 .auth()
5105 .signing_keys()
5106 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
5107 .into_iter()
5108 .find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
5109 .ok_or_else(|| {
5110 ControlError::InvalidArgument(format!(
5111 "{} is not registered on this server, so nobody reading this event later \
5112 could check its signature",
5113 signed.key_fingerprint
5114 ))
5115 })?;
5116 if key.principal_id != signed.event.actor {
5123 return Err(ControlError::InvalidArgument(format!(
5124 "this event is attributed to {:?} and is signed by {}, which is {:?}'s key; \
5125 an approval is the actor's own word, so it must carry the actor's own \
5126 signature",
5127 signed.event.actor, signed.key_fingerprint, key.principal_id
5128 )));
5129 }
5130 let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
5131 .map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
5132 gunnar_core::forge::verify_event(&signed, &public)
5133 .map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
5134
5135 let (namespace, name) = signed.event.repo.split_once('/').ok_or_else(|| {
5141 ControlError::InvalidArgument(format!(
5142 "`repo` must be `<namespace>/<repo>`, got {:?}",
5143 signed.event.repo
5144 ))
5145 })?;
5146 let path = RepoPath::new(namespace, name);
5147 match &signed.event.event {
5148 gunnar_core::Event::IssueOpened { .. } => {}
5151 gunnar_core::Event::IssueCommented { .. } | gunnar_core::Event::IssueClosed { .. } => {
5155 let repo_events: Vec<gunnar_core::ForgeEvent> = self
5156 .forge_events()?
5157 .into_iter()
5158 .filter(|e| e.event.repo == signed.event.repo)
5159 .map(|e| e.event)
5160 .collect();
5161 match gunnar_core::forge::fold_issue(signed.event.pull, &repo_events) {
5162 None => {
5163 return Err(ControlError::FailedPrecondition(format!(
5164 "no issue #{} in {}: nobody has opened one, so an event on it is \
5165 evidence about nothing",
5166 signed.event.pull, signed.event.repo
5167 )));
5168 }
5169 Some(issue) if !issue.open => {
5170 return Err(ControlError::FailedPrecondition(format!(
5171 "issue #{} in {} is closed; an event on it cannot change what has \
5172 already happened",
5173 signed.event.pull, signed.event.repo
5174 )));
5175 }
5176 Some(_) => {}
5177 }
5178 }
5179 gunnar_core::Event::Commented { .. }
5182 | gunnar_core::Event::Approved { .. }
5183 | gunnar_core::Event::Rejected { .. } => {
5184 let record = self.get_pull(&path, signed.event.pull)?;
5185 if record.state != gunnar_proto::PullRequestState::Open as i32 {
5186 return Err(ControlError::FailedPrecondition(format!(
5187 "pull request #{} in {} is not open; a review event cannot change what \
5188 has already happened",
5189 signed.event.pull, signed.event.repo
5190 )));
5191 }
5192 }
5193 }
5194
5195 let log = self.stores.forge_log();
5196 let mut line = serde_json::to_string(&signed)
5197 .map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
5198 line.push('\n');
5199 let _framed = gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
5200 .map_err(|err| ControlError::Unavailable(format!("framing the line in the transaction log: {err}")))?;
5201 use std::io::Write as _;
5202 let mut file = std::fs::OpenOptions::new()
5203 .create(true)
5204 .append(true)
5205 .open(log.records_path())
5206 .map_err(|err| ControlError::Unavailable(format!("opening the forge log: {err}")))?;
5207 file.write_all(line.as_bytes()).map_err(|err| {
5208 ControlError::Unavailable(format!("appending to the forge log: {err}"))
5209 })?;
5210 let count = self
5213 .stores
5214 .forge_log()
5215 .lines_of()
5216 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
5217 .len() as u64;
5218 Ok(count.saturating_sub(1))
5219 }
5220
5221 fn forge_review(
5222 &self,
5223 repo: &str,
5224 pull: u64,
5225 ) -> ControlResult<(
5226 gunnar_proto::ReviewState,
5227 Vec<gunnar_proto::SignedForgeEvent>,
5228 )> {
5229 let all = self.forge_events()?;
5230 let mine: Vec<gunnar_core::ForgeEvent> = all
5231 .iter()
5232 .filter(|signed| signed.event.repo == repo)
5233 .map(|signed| signed.event.clone())
5234 .collect();
5235 let state = gunnar_core::forge::fold(pull, &mine);
5236 let events = all
5237 .iter()
5238 .filter(|signed| signed.event.repo == repo && signed.event.pull == pull)
5239 .cloned()
5240 .map(to_proto_event)
5241 .collect();
5242 Ok((self.review_state(repo, &state, &all)?, events))
5243 }
5244
5245 fn forge_reviews(&self, repo: &str) -> ControlResult<Vec<gunnar_proto::ReviewState>> {
5246 let all = self.forge_events()?;
5247 let mine: Vec<gunnar_core::ForgeEvent> = all
5248 .iter()
5249 .filter(|signed| signed.event.repo == repo)
5250 .map(|signed| signed.event.clone())
5251 .collect();
5252 let mut numbers: Vec<u64> = mine.iter().map(|event| event.pull).collect();
5253 numbers.sort_unstable();
5254 numbers.dedup();
5255 let mut out = Vec::new();
5256 for pull in numbers {
5257 let state = gunnar_core::forge::fold(pull, &mine);
5258 out.push(self.review_state(repo, &state, &all)?);
5259 }
5260 Ok(out)
5261 }
5262
5263 fn forge_issue(
5264 &self,
5265 repo: &str,
5266 number: u64,
5267 ) -> ControlResult<Option<(gunnar_proto::IssueState, Vec<gunnar_proto::SignedForgeEvent>)>> {
5268 let all = self.forge_events()?;
5269 let mine: Vec<gunnar_core::ForgeEvent> = all
5270 .iter()
5271 .filter(|signed| signed.event.repo == repo)
5272 .map(|signed| signed.event.clone())
5273 .collect();
5274 let Some(issue) = gunnar_core::forge::fold_issue(number, &mine) else {
5275 return Ok(None);
5276 };
5277 let mut opened_seen = 0u64;
5282 let events = all
5283 .iter()
5284 .filter(|signed| {
5285 if signed.event.repo != repo {
5286 return false;
5287 }
5288 match &signed.event.event {
5289 gunnar_core::Event::IssueOpened { .. } => {
5290 opened_seen += 1;
5291 opened_seen == number
5292 }
5293 gunnar_core::Event::IssueCommented { .. }
5294 | gunnar_core::Event::IssueClosed { .. } => signed.event.pull == number,
5295 _ => false,
5296 }
5297 })
5298 .cloned()
5299 .map(to_proto_event)
5300 .collect();
5301 Ok(Some((issue_state(repo, &issue), events)))
5302 }
5303
5304 fn forge_issues(&self, repo: &str) -> ControlResult<Vec<gunnar_proto::IssueState>> {
5305 let mine: Vec<gunnar_core::ForgeEvent> = self
5306 .forge_events()?
5307 .into_iter()
5308 .filter(|signed| signed.event.repo == repo)
5309 .map(|signed| signed.event)
5310 .collect();
5311 Ok(gunnar_core::forge::fold_issues(&mine)
5312 .iter()
5313 .map(|issue| issue_state(repo, issue))
5314 .collect())
5315 }
5316
5317 fn stats_extras(&self) -> ControlResult<gunnar_grpc::seam::StatsExtras> {
5327 let tombstones = self.stores.tombstones().unwrap_or_default();
5328 let report = self.replication.report();
5329 #[allow(unused_mut)]
5332 let mut tenants = crate::dashboard::TenantCounts::default();
5333 #[cfg(feature = "metered")]
5334 if let Some(slot) = &self.entitlement {
5335 tenants = slot.counts();
5336 }
5337 Ok(gunnar_grpc::seam::StatsExtras {
5338 tombstoned_repos: tombstones.len() as u64,
5339 tombstoned_bytes: tombstones.iter().map(|row| row.bytes).sum(),
5340 replication_configured: report.configured,
5341 replication_enqueued: report.enqueued,
5342 replication_synced: report.synced,
5343 replication_failed: report.failed,
5344 replication_bus_gaps: report.bus_gaps,
5345 replication_refs_total: report.refs.total,
5346 replication_refs_unmirrored: report.refs.unmirrored,
5347 standbys: report
5348 .standbys
5349 .into_iter()
5350 .map(|row| gunnar_proto::StandbyStatus {
5351 verdict: row.verdict().word().to_owned(),
5355 name: row.name,
5356 remote: row.remote,
5357 pending: row.pending,
5358 consecutive_failures: row.consecutive_failures,
5359 last_error: row.last_error,
5360 last_success_unix_ms: row.last_success_unix_ms,
5361 needs_full_resync: row.needs_full_resync,
5362 stale: row.stale,
5363 })
5364 .collect(),
5365 tenants_paid: tenants.paid,
5366 tenants_grace: tenants.grace,
5367 tenants_suspended: tenants.suspended,
5368 })
5369 }
5370
5371 fn counts(&self) -> ControlResult<ControlCounts> {
5372 let (namespaces, repos) = self.stores.catalog().counts().map_err(internal)?;
5373 let (principals, keys, grants) = self.stores.auth().counts().map_err(internal)?;
5374 Ok(ControlCounts {
5375 namespaces: namespaces as u64,
5376 repos: repos as u64,
5377 principals: principals as u64,
5378 keys: keys as u64,
5379 grants: grants as u64,
5380 })
5381 }
5382
5383 fn verify_chain(
5384 &self,
5385 request: &gunnar_proto::VerifyChainRequest,
5386 ) -> ControlResult<gunnar_proto::VerifyChainResponse> {
5387 let target = match request.target.as_str() {
5391 "" => None,
5392 spec => Some(spec),
5393 };
5394 let chain = if !request.artifact_sha256.is_empty() {
5395 crate::verify::verify_artifact(&self.stores, self, &request.artifact_sha256, target)
5396 } else {
5397 let Some(repo) = &request.repo else {
5398 return Err(ControlError::InvalidArgument(
5399 "name a repository, or an artifact_sha256 to start from".to_owned(),
5400 ));
5401 };
5402 let rev = match request.rev.as_str() {
5403 "" => "HEAD",
5404 rev => rev,
5405 };
5406 crate::verify::verify(&self.stores, self, repo, rev, target)
5407 }
5408 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
5413 Ok(chain.to_proto())
5414 }
5415
5416 fn blame(
5417 &self,
5418 path: &RepoPath,
5419 rev: &str,
5420 file: &str,
5421 ) -> ControlResult<gunnar_proto::BlameResponse> {
5422 let rev = if rev.is_empty() { "HEAD" } else { rev };
5429 let rows = crate::blame::blame(&self.stores, self, path, rev, file)
5430 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
5431 Ok(gunnar_proto::BlameResponse {
5432 lines: rows
5433 .into_iter()
5434 .map(|row| gunnar_proto::BlameLine {
5435 commit: row.commit.to_hex().to_string(),
5436 author: row.author,
5437 line: row.line,
5438 })
5439 .collect(),
5440 })
5441 }
5442
5443 fn search(
5444 &self,
5445 path: &RepoPath,
5446 rev: &str,
5447 pattern: &str,
5448 max_file_bytes: u64,
5449 max_matches: u32,
5450 ) -> ControlResult<gunnar_proto::SearchResponse> {
5451 let rev = if rev.is_empty() { "HEAD" } else { rev };
5455 let defaults = crate::search::Limits::default();
5456 let limits = crate::search::Limits {
5457 max_file_bytes: if max_file_bytes == 0 {
5458 defaults.max_file_bytes
5459 } else {
5460 max_file_bytes
5461 },
5462 max_matches: if max_matches == 0 {
5463 defaults.max_matches
5464 } else {
5465 max_matches as usize
5466 },
5467 };
5468 let hits = crate::search::search(&self.stores, self, path, rev, pattern, limits)
5469 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
5470 Ok(gunnar_proto::SearchResponse {
5471 matches: hits
5472 .into_iter()
5473 .map(|hit| gunnar_proto::SearchMatch {
5474 path: hit.path,
5475 line_number: hit.line_number as u32,
5476 line: hit.line,
5477 })
5478 .collect(),
5479 })
5480 }
5481
5482 fn issuer_info(&self) -> ControlResult<gunnar_proto::IssuerInfoResponse> {
5496 let pki = self.stores.pki_readonly().map_err(|err| {
5497 ControlError::Unavailable(format!("the PKI document could not be opened: {err:#}"))
5498 })?;
5499 crate::admin::issuer_document(&pki).map_err(|err| {
5503 ControlError::Unavailable(format!("the PKI document could not be read: {err:#}"))
5504 })
5505 }
5506
5507 fn setting(&self, name: &str) -> ControlResult<Option<Vec<u8>>> {
5508 self.stores.settings().get(name).map_err(internal)
5509 }
5510
5511 fn set_setting(&self, name: &str, value: &[u8]) -> ControlResult<Option<Vec<u8>>> {
5512 self.stores.settings().set(name, value).map_err(internal)
5513 }
5514
5515 fn audit_trail(&self, at_seq: u64) -> ControlResult<gunnar_proto::AuditTrailResponse> {
5531 let path = self.stores.audit_path();
5532 let replicated = !self.stores.is_twin()
5537 && !crate::twin::is_restoring(self.stores.root())
5538 && gunnar_core::txlog::sink::log_for(&path)
5539 .is_some_and(|(_, rel)| rel == gunnar_core::txlog::AUDIT_REL);
5540 let mut out = gunnar_proto::AuditTrailResponse {
5541 present: false,
5542 entries: 0,
5543 tip: String::new(),
5544 verified: false,
5545 replicated,
5546 hash_at_seq: String::new(),
5547 why: String::new(),
5548 tip_seq: 0,
5549 };
5550 let chain = match gunnar_core::AuditChain::open(&path) {
5551 Ok(chain) => chain,
5552 Err(err) => {
5553 out.why = format!("{err}");
5554 return Ok(out);
5555 }
5556 };
5557 let entries = match chain.entries() {
5558 Ok(entries) => entries,
5559 Err(err) => {
5560 out.why = format!("{err}");
5561 return Ok(out);
5562 }
5563 };
5564 out.present = true;
5565 out.entries = entries.len() as u64;
5566 if let Some(last) = entries.last() {
5567 out.tip.clone_from(&last.hash);
5568 out.tip_seq = last.seq;
5570 }
5571 match chain.verify() {
5572 Ok(_) => out.verified = true,
5573 Err(err) => out.why = format!("{err}"),
5574 }
5575 if at_seq != 0 {
5576 match entries.iter().find(|e| e.seq == at_seq) {
5581 Some(e) => out.hash_at_seq = e.hash.clone(),
5582 None => {
5583 let sep = if out.why.is_empty() { "" } else { "; " };
5584 out.why = format!(
5585 "{}{sep}this trail holds no entry at seq {at_seq} (it holds {} entries)",
5586 out.why,
5587 entries.len()
5588 );
5589 }
5590 }
5591 }
5592 Ok(out)
5593 }
5594
5595 fn storage_info(&self, path: Option<&RepoPath>) -> ControlResult<StorageInfoResponse> {
5596 let repos = match path {
5597 Some(path) => vec![self.open(path)?.0],
5598 None => self.stores.catalog().list(None).map_err(internal)?,
5599 };
5600 let mut total_hot = 0u64;
5601 let mut total_explode = 0u64;
5602 let mut total_budget = 0u64;
5603 let bands: std::collections::HashMap<std::path::PathBuf, crate::storage::ExplodeBand> =
5609 self.stores.engine().explode_use().into_iter().collect();
5610 let mut out = Vec::new();
5611 for repo in repos {
5612 let store = self.stores.repo_store(&repo).map_err(|err| {
5613 ControlError::Internal(format!("opening the repository's stores: {err:#}"))
5614 })?;
5615 let (loose, _objects_dir) = object_dir_stats(store.git_dir());
5616 let bytes = repo_bytes(store.git_dir()).hot();
5624 let refs = store.serve().refs().map_err(store_error)?.len() as u64;
5625 total_hot += bytes;
5626 let band = bands.get(store.git_dir()).copied().unwrap_or_default();
5630 total_explode += band.used();
5631 total_budget += band.budget;
5632 out.push(RepoStorage {
5633 repo: Some(RepoPath::new(&repo.namespace, &repo.name)),
5634 hash_kind: hash_kind(repo.hash_kind) as i32,
5635 loose_object_count: loose,
5636 hot_bytes: bytes,
5637 cold_bytes: 0,
5643 cold_object_count: 0,
5644 cold_measured: false,
5645 last_repack_at_unix_ms: 0,
5646 ref_count: refs,
5647 explode_bytes: band.used(),
5648 explode_budget_bytes: band.budget,
5649 explode_evicted_bytes: band.evicted_bytes,
5650 });
5651 }
5652 let engine = self.stores.engine();
5662 let (object_backend, ref_backend) =
5663 (engine.describe(), engine.describe());
5664 let tombstones = self.stores.tombstones().unwrap_or_default();
5669 let total_tombstoned_bytes = tombstones.iter().map(|row| row.bytes).sum();
5670 let tombstoned = tombstones
5671 .into_iter()
5672 .map(|row| gunnar_proto::TombstonedRepo {
5673 repo: Some(RepoPath {
5674 namespace: row.path.namespace().to_owned(),
5675 repo: row.path.name().to_owned(),
5676 }),
5677 bytes: row.bytes,
5678 deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
5679 purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
5680 })
5681 .collect();
5682 Ok(StorageInfoResponse {
5683 object_backend,
5684 ref_backend,
5685 engine: Some(engine.report()),
5687 total_hot_bytes: total_hot,
5688 total_cold_bytes: 0,
5690 cold_measured: false,
5691 repos: out,
5692 total_explode_bytes: total_explode,
5693 total_explode_budget_bytes: total_budget,
5694 total_tombstoned_bytes,
5695 tombstoned,
5696 reality: Some(self.stores.storage_reality()),
5701 })
5702 }
5703
5704 fn capacity(
5730 &self,
5731 request: &gunnar_proto::CapacityRequest,
5732 ) -> ControlResult<gunnar_proto::CapacityResponse> {
5733 let Some(gate) = self.headroom.as_ref() else {
5734 return Err(ControlError::Unimplemented(
5735 "Admin.Capacity (no disk-headroom gate is wired on this control plane, so the \
5736 --disk-floor in force is unknown and no quote can be made against it)",
5737 ));
5738 };
5739 let held_bytes = if request.namespace.is_empty() {
5744 0
5745 } else {
5746 self.usage(Some(&request.namespace))?
5747 .first()
5748 .map(|row| row.pack_bytes)
5749 .unwrap_or(0)
5750 };
5751 let root = gate.root().to_path_buf();
5752 let mut answer = crate::capacity::answer(&crate::capacity::Question {
5753 room: gate.headroom().map_err(|err| err.to_string()),
5754 floor_spec: gate.floor().to_string(),
5755 requested_bytes: request.requested_bytes,
5756 held_bytes,
5757 namespace: request.namespace.clone(),
5758 growth: crate::capacity::GrowthFacts::measure(&root),
5759 machine: crate::capacity::MachineFacts::measure(),
5760 root: root.clone(),
5761 });
5762 if request.disks {
5765 let slow = self.stores.slow();
5766 let serialdat = std::path::PathBuf::from(slow.serialdat());
5767 let mut disks = crate::capacity::measure_disks(&root, Some(&serialdat), gate.floor());
5768 let log = slow.path_of(crate::slow::LOG_DIR);
5772 let known: Vec<(&str, std::path::PathBuf)> = [
5773 ("metadata copy", slow.path_of(crate::slow::META_DIR)),
5774 ("crash record", slow.path_of(crate::slow::CRASH_DIR)),
5775 ("audit trail", log.join("audit-chain.jsonl")),
5776 ("logs", log),
5777 ("grow journal", serialdat.join("grow-journal")),
5778 ("repository mirror", serialdat.join(crate::slow::REPOS_DIR)),
5779 ]
5780 .into_iter()
5781 .filter(|(_, path)| path.starts_with(&serialdat))
5782 .collect();
5783 let used = disks
5784 .mounts
5785 .iter()
5786 .find(|m| m.labels.iter().any(|l| l == "serialdat") && m.measured)
5787 .map(|m| m.used_bytes);
5788 if disks.mounts.iter().any(|m| m.labels.iter().any(|l| l == "serialdat")) {
5789 disks.serialdat = crate::capacity::serialdat_breakdown(&known, used);
5790 }
5791 answer.disks = Some(disks);
5792 }
5793 Ok(answer)
5794 }
5795
5796 fn grow_start(
5799 &self,
5800 request: &gunnar_proto::GrowStartRequest,
5801 ) -> ControlResult<gunnar_proto::GrowStatusResponse> {
5802 self.grow_machine()?
5803 .start(request)
5804 .map_err(ControlError::FailedPrecondition)
5805 }
5806
5807 fn grow_status(&self) -> ControlResult<gunnar_proto::GrowStatusResponse> {
5808 match &self.grow {
5809 Some(grow) => Ok(grow.status()),
5810 None => Ok(gunnar_proto::GrowStatusResponse {
5814 phase: "idle".to_owned(),
5815 engine_present: false,
5816 serving: true,
5817 server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
5818 detail: "this control plane has no runtime growth wired in".to_owned(),
5819 hotplug: crate::dashboard::probe::pci_hotplug().present,
5820 journal: gunnar_grpc::grow_journal::Summary::measure().map(|s| s.to_wire()),
5821 ..Default::default()
5822 }),
5823 }
5824 }
5825
5826 fn grow_resume(&self, give_up: bool) -> ControlResult<gunnar_proto::GrowStatusResponse> {
5827 self.grow_machine()?
5828 .resume(give_up)
5829 .map_err(ControlError::FailedPrecondition)
5830 }
5831
5832 fn usage(&self, namespace: Option<&str>) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
5848 self.usage_readable()?;
5853 if let Some(name) = namespace {
5861 let measured_at_unix_ms = gunnar_core::now_ms();
5862 let Some((held, _)) = self.stores.usage_ledger().exact(&self.stores, name).map_err(internal)? else {
5863 return Err(ControlError::NotFound(format!(
5864 "namespace {name} is not in the catalog"
5865 )));
5866 };
5867 let objects = self.open_objects_of(name);
5868 return Ok(vec![self.usage_row(name, &held, objects, measured_at_unix_ms)]);
5869 }
5870 self.usage_all()
5871 }
5872
5873 fn repo_usage(&self, namespace: &str) -> ControlResult<Vec<gunnar_proto::RepoStored>> {
5876 let Some((_, repos)) = self.stores.usage_ledger().exact(&self.stores, namespace).map_err(internal)? else {
5877 return Err(ControlError::NotFound(format!(
5878 "namespace {namespace} is not in the catalog"
5879 )));
5880 };
5881 let open_objects: std::collections::HashMap<std::path::PathBuf, u64> =
5882 self.stores.engine().open_object_counts().into_iter().collect();
5883 Ok(repos
5885 .into_iter()
5886 .map(|(name, bytes, dir)| gunnar_proto::RepoStored {
5887 repo: Some(gunnar_proto::RepoPath::new(namespace, &name)),
5888 stored_bytes: bytes.pack,
5889 cache_bytes: bytes.cache,
5890 lfs_bytes: bytes.lfs,
5891 objects: dir.and_then(|d| open_objects.get(&d).copied()).unwrap_or(0),
5892 })
5893 .collect())
5894 }
5895
5896 fn usage_page(
5901 &self,
5902 page_token: &str,
5903 page_size: usize,
5904 changed_since: &str,
5905 ) -> ControlResult<gunnar_proto::ReadManyUsageResponse> {
5906 self.usage_readable()?;
5907 let now = gunnar_core::now_ms();
5908 let ledger = self.stores.usage_ledger();
5909 let (cursor, after) = gunnar_grpc::seam::usage_page_token::decode(page_token);
5910 let cursor = if cursor.is_empty() { ledger.cursor_now(now) } else { cursor };
5914 let (rows, more, changed_only) = ledger
5915 .page(&self.stores, &after, page_size, changed_since, now)
5916 .map_err(internal)?;
5917 let next_page_token = match (more, rows.last()) {
5918 (true, Some((last, _))) => gunnar_grpc::seam::usage_page_token::encode(&cursor, last),
5919 _ => String::new(),
5920 };
5921 let usages = rows
5922 .iter()
5923 .map(|(name, held)| self.usage_row(name, held, held.objects, now))
5924 .collect();
5925 Ok(gunnar_proto::ReadManyUsageResponse {
5926 usages,
5927 next_page_token,
5928 cursor,
5929 changed_only,
5930 as_of_unix_ms: now,
5931 })
5932 }
5933
5934 fn activity(
5954 &self,
5955 namespace: &str,
5956 since_unix_ms: i64,
5957 until_unix_ms: i64,
5958 ) -> ControlResult<(Vec<gunnar_proto::DayActivity>, i64)> {
5959 if self
5960 .stores
5961 .catalog()
5962 .namespace(namespace)
5963 .map_err(internal)?
5964 .is_none()
5965 {
5966 return Err(ControlError::NotFound(format!(
5967 "namespace {namespace} is not in the catalog"
5968 )));
5969 }
5970 let now = gunnar_core::now_ms();
5971 let flow = self.stores.flow();
5972 if let Err(err) = flow.flush(now) {
5973 tracing::warn!(
5977 target: "gunnar::store",
5978 event = "flow.flush_failed",
5979 %err,
5980 "could not flush the flow ledger before an Activity read; the series is as of \
5981 the last successful flush"
5982 );
5983 }
5984 let today = gunnar_core::day_of(now);
5985 let from = if since_unix_ms == 0 {
5986 today - gunnar_core::RETENTION_DAYS
5987 } else {
5988 gunnar_core::day_of(since_unix_ms)
5989 }
5990 .max(today - gunnar_core::RETENTION_DAYS);
5991 let to = if until_unix_ms == 0 {
5992 today
5993 } else {
5994 gunnar_core::day_of(until_unix_ms)
5995 }
5996 .min(today);
5997 if from > to {
5998 return Ok((Vec::new(), gunnar_core::RETENTION_DAYS));
5999 }
6000 let days = flow
6001 .store()
6002 .series(namespace, from, to)
6003 .map_err(internal)?
6004 .into_iter()
6005 .map(|(day, row)| day_activity(day, &row))
6006 .collect();
6007 Ok((days, gunnar_core::RETENTION_DAYS))
6008 }
6009
6010 fn admit_plan(&self, namespace: &str) -> ControlResult<gunnar_proto::AdmitPlanResponse> {
6016 use gunnar_ssh::edda_admit::{Consumer, Work};
6017 use gunnar_ssh::governor::OpClass;
6018 if self
6019 .stores
6020 .catalog()
6021 .namespace(namespace)
6022 .map_err(internal)?
6023 .is_none()
6024 {
6025 return Err(ControlError::NotFound(format!(
6026 "namespace {namespace} is not in the catalog"
6027 )));
6028 }
6029 let Some(gate) = &self.admission else {
6030 return Err(ControlError::Unimplemented("Usage.AdmitPlan"));
6031 };
6032 let policy = gate.admit_policy();
6033 let usage = gate.admit_usage(&*policy, namespace);
6034 let limit = policy.limit();
6035 let who = Consumer::Namespace(namespace.to_owned());
6036 let estimates = policy.estimates();
6037 let kinds = [OpClass::Clone, OpClass::Fetch, OpClass::Push, OpClass::LsRemote]
6038 .into_iter()
6039 .map(|class| {
6040 let kind = gunnar_ssh::admission::work_kind(class);
6041 let plan = policy.plan(&Work::new(kind, None), &who, &usage);
6042 gunnar_proto::AdmitPlanKind {
6043 kind: kind.name().to_owned(),
6044 admit_class: plan.class.name().to_owned(),
6045 why: plan.why,
6046 estimate_bytes: plan.estimate.usage_bytes(),
6047 estimate_cpu_ms: plan.estimate.cpu_ms,
6048 samples: estimates
6049 .iter()
6050 .find(|row| row.kind == kind)
6051 .map_or(0, |row| row.samples),
6052 }
6053 })
6054 .collect();
6055 Ok(gunnar_proto::AdmitPlanResponse {
6056 policy: policy.policy().to_owned(),
6057 window_days: usage.window_days,
6058 used_bytes: usage.bytes,
6059 limit_bytes: limit.map(|l| l.bytes),
6060 budget_left_bytes: limit.map(|l| l.bytes.saturating_sub(usage.bytes)),
6061 kinds,
6062 })
6063 }
6064
6065 #[cfg(feature = "metered")]
6068 fn entitlement_get(
6069 &self,
6070 namespace: &str,
6071 ) -> ControlResult<Option<gunnar_proto::EntitlementFact>> {
6072 self.entitlement()?
6073 .get(namespace)
6074 .map_err(|err| ControlError::Internal(format!("{err:#}")))
6075 }
6076
6077 #[cfg(feature = "metered")]
6078 fn entitlement_list(&self) -> ControlResult<Vec<gunnar_proto::EntitlementFact>> {
6079 self.entitlement()?
6080 .list()
6081 .map_err(|err| ControlError::Internal(format!("{err:#}")))
6082 }
6083
6084 #[cfg(feature = "metered")]
6085 fn entitlement_refused_at_boot(&self) -> ControlResult<Option<gunnar_proto::RefusedAtBoot>> {
6086 Ok(self.entitlement()?.refusal())
6087 }
6088
6089 #[cfg(feature = "metered")]
6095 fn entitlement_set(
6096 &self,
6097 fact: gunnar_proto::EntitlementFact,
6098 ) -> ControlResult<(gunnar_proto::EntitlementState, u64)> {
6099 use crate::entitlement::SetRefused;
6100 let (state, seq) = self.entitlement()?.set(fact).map_err(|err| match err {
6101 SetRefused::Forged { .. } => ControlError::PermissionDenied(err.to_string()),
6102 SetRefused::Replayed { .. } | SetRefused::Unstamped { .. } | SetRefused::Behind { .. } => ControlError::PermissionDenied(err.to_string()),
6111 SetRefused::Malformed(_) => ControlError::InvalidArgument(err.to_string()),
6112 SetRefused::NotRecorded { .. } => ControlError::FailedPrecondition(err.to_string()),
6113 SetRefused::NoKey { .. } => ControlError::FailedPrecondition(err.to_string()),
6114 })?;
6115 Ok((crate::entitlement::state_to_proto(state), seq))
6116 }
6117
6118 #[cfg(feature = "metered")]
6130 fn actor_ticket_public_key_hex(&self) -> ControlResult<String> {
6131 Ok(self
6132 .stores
6133 .actor_ticket_key()
6134 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
6135 .public_key_hex()
6136 .to_owned())
6137 }
6138
6139 #[cfg(feature = "metered")]
6140 fn actor_ticket_sign(&self, message: &[u8]) -> ControlResult<[u8; 64]> {
6141 Ok(self
6142 .stores
6143 .actor_ticket_key()
6144 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
6145 .sign(message))
6146 }
6147
6148 #[cfg(feature = "metered")]
6152 fn sealed_get(&self, name: &str) -> ControlResult<Option<(String, Vec<u8>)>> {
6153 self.stores.sealed_slot_get(name).map_err(|err| ControlError::Internal(format!("{err:#}")))
6154 }
6155
6156 #[cfg(feature = "metered")]
6157 fn sealed_put(&self, name: &str, owner: &str, value: &[u8]) -> ControlResult<bool> {
6158 self.stores.sealed_slot_put(name, owner, value).map_err(|err| ControlError::Unavailable(format!("{err:#}")))
6159 }
6160
6161 fn mirror_add(
6170 &self,
6171 path: &RepoPath,
6172 remote_url: &str,
6173 prune: bool,
6174 create_repo: bool,
6175 ) -> ControlResult<MirrorConfig> {
6176 if remote_url.trim().is_empty() {
6177 return Err(ControlError::InvalidArgument(
6178 "a mirror needs a remote URL".to_owned(),
6179 ));
6180 }
6181 crate::mirror_fetch::classify(remote_url).map_err(ControlError::InvalidArgument)?;
6190 crate::mirror_fetch::check_destination(remote_url, self.stores.root())
6191 .map_err(ControlError::InvalidArgument)?;
6192 let repo_ref = Self::repo_ref(path)?;
6193 let creates = create_repo && self.get_repo(path).is_err();
6203 if creates {
6204 if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
6207 return Err(ControlError::ResourceExhausted(line));
6208 }
6209 if self.stores.catalog().namespace(path.namespace.as_str()).map_err(internal)?.is_none() {
6210 return Err(ControlError::NotFound(format!("namespace {}", path.namespace)));
6211 }
6212 } else {
6213 self.open(path)?;
6217 }
6218 let mut record = gunnar_core::MirrorRecord::new(
6219 &repo_ref,
6220 remote_url,
6221 prune,
6222 gunnar_grpc::bus::now_unix_ms(),
6223 );
6224 record.creates_repo = creates;
6225 let stored = self
6226 .stores
6227 .mirrors()
6228 .set(&repo_ref, record)
6229 .map_err(internal)?;
6230 Ok(mirror_out(&stored))
6231 }
6232
6233 fn mirror_sync(&self, path: &RepoPath) -> ControlResult<MirrorConfig> {
6263 let record = self.mirror_record(path)?;
6264 let repo_ref = Self::repo_ref(path)?;
6265 crate::mirror_fetch::classify(&record.remote_url).map_err(ControlError::InvalidArgument)?;
6270 crate::mirror_fetch::check_destination(&record.remote_url, self.stores.root())
6273 .map_err(ControlError::InvalidArgument)?;
6274 let job = self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
6275 path: repo_ref.key(),
6276 upstream: record.remote_url.clone(),
6277 prune: record.prune,
6278 });
6279 let mut out = mirror_out(&record);
6280 out.job = Some(mirror_job_out(path, job));
6281 Ok(out)
6282 }
6283
6284 fn mirror_set_credential(
6297 &self,
6298 path: &RepoPath,
6299 kind: gunnar_proto::MirrorCredentialKind,
6300 secret: &str,
6301 username: &str,
6302 ) -> ControlResult<MirrorConfig> {
6303 gunnar_grpc::seam::credential_arguments(kind, secret)?;
6304 self.mirror_record(path)?;
6305 let repo_ref = Self::repo_ref(path)?;
6306
6307 let kind = credential_kind_in(kind);
6308 let secrets = self.stores.secrets().map_err(|err| {
6309 ControlError::Internal(format!("the sealed secret store is not available: {err:#}"))
6310 })?;
6311 let key = gunnar_core::credential_secret_key(&repo_ref);
6312 let stored = if kind.is_set() {
6313 gunnar_store::SecretStore::put(&secrets, &key, secret.as_bytes())
6314 } else {
6315 gunnar_store::SecretStore::delete(&secrets, &key)
6316 };
6317 stored.map_err(|err| {
6318 ControlError::Internal(format!(
6320 "the mirror credential for {path} could not be sealed: {err}"
6321 ))
6322 })?;
6323
6324 let record = self
6325 .stores
6326 .mirrors()
6327 .set_credential(&repo_ref, kind, username)
6328 .map_err(internal)?
6329 .ok_or_else(|| ControlError::NotFound(format!("no mirror registered for {path}")))?;
6330 Ok(mirror_out(&record))
6331 }
6332
6333 fn mirror_status(
6334 &self,
6335 namespace: Option<&str>,
6336 path: Option<&RepoPath>,
6337 ) -> ControlResult<Vec<MirrorConfig>> {
6338 let all = self.stores.mirrors().all().map_err(internal)?;
6339 Ok(all
6340 .iter()
6341 .filter(|record| match path {
6342 Some(want) => record.repo == format!("{}/{}", want.namespace, want.repo),
6343 None => true,
6344 })
6345 .filter(|record| match namespace {
6346 Some(want) => record
6347 .repo
6348 .split_once('/')
6349 .is_some_and(|(ns, _)| ns == want),
6350 None => true,
6351 })
6352 .map(|record| {
6353 let mut out = mirror_out(record);
6354 out.job = out
6360 .repo
6361 .as_ref()
6362 .zip(self.mirror.status(&record.repo))
6363 .map(|(path, job)| mirror_job_out(path, job));
6364 if out.job.is_none() && !record.last_error.is_empty() {
6371 out.job = out.repo.clone().map(|path| gunnar_proto::MirrorJob {
6372 job_id: String::new(),
6373 path: Some(path),
6374 state: gunnar_proto::MirrorJobState::Failed as i32,
6375 error: record.last_error.clone(),
6376 ..Default::default()
6377 });
6378 }
6379 out
6380 })
6381 .collect())
6382 }
6383
6384 fn mirror_remove(&self, path: &RepoPath) -> ControlResult<()> {
6405 self.mirror_record(path)?;
6406 let repo_ref = Self::repo_ref(path)?;
6407 if !self.stores.mirrors().remove(&repo_ref).map_err(internal)? {
6408 return Err(ControlError::NotFound(format!(
6409 "no mirror registered for {path}"
6410 )));
6411 }
6412 let secrets = self.stores.secrets().map_err(|err| {
6413 ControlError::Internal(format!("the sealed secret store is not available: {err:#}"))
6414 })?;
6415 gunnar_store::SecretStore::delete(&secrets, &gunnar_core::credential_secret_key(&repo_ref))
6416 .map_err(|err| {
6417 ControlError::Internal(format!(
6421 "the registration for {path} was removed and its sealed credential could \
6422 not be deleted: {err}. The secret is still on this box; remove it before \
6423 trusting that it is gone."
6424 ))
6425 })?;
6426 Ok(())
6427 }
6428
6429 #[cfg(feature = "replication")]
6432 fn replication_set(
6433 &self,
6434 target: gunnar_proto::ReplicationTarget,
6435 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
6436 let slot = self.replication_slot()?;
6437 let mut target = crate::replication_wire::target_from_proto(target)?;
6438 let minted_public = if target.identity.is_none() {
6448 let (path, public) = crate::replication_wire::ensure_identity(self.stores.root())?;
6449 target.identity = Some(path);
6450 Some(public)
6451 } else {
6452 None
6453 };
6454 crate::replication_wire::require_public_only_identity(&target)?;
6460 let mut config = slot
6461 .config()
6462 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6463 config.set_target(target);
6467 slot.apply(config)
6468 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6469 let mut out = self.replication_snapshot(false)?;
6470 out.identity_public_key = minted_public.unwrap_or_default();
6471 Ok(out)
6472 }
6473
6474 #[cfg(feature = "replication")]
6475 fn replication_remove(
6476 &self,
6477 name: &str,
6478 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
6479 let slot = self.replication_slot()?;
6480 let mut config = slot
6481 .config()
6482 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6483 if !config.remove_target(name) {
6484 return Err(ControlError::NotFound(format!(
6489 "no replication target named {name:?}"
6490 )));
6491 }
6492 slot.apply(config)
6493 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6494 self.replication_snapshot(false)
6495 }
6496
6497 fn is_group_member(&self, principal_id: &str, group: &str) -> ControlResult<bool> {
6500 self.stores
6501 .auth()
6502 .groups_of(principal_id)
6503 .map(|groups| groups.iter().any(|g| g == group))
6504 .map_err(|err| ControlError::Unavailable(format!("reading group membership: {err}")))
6505 }
6506
6507 fn become_twin(&self, primary: &str) -> ControlResult<String> {
6508 let key = crate::twin::become_twin(&self.stores, primary)
6509 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
6510 crate::twin::start_follower(Arc::clone(&self.stores));
6513 Ok(key)
6514 }
6515
6516 #[cfg(feature = "wg-client")]
6517 fn assign_tunnel_address(&self, address: &str) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
6518 let window_open = self
6519 .stores
6520 .auth()
6521 .bootstrap_window_open()
6522 .map_err(|err| ControlError::Unavailable(format!("reading the bootstrap window: {err}")))?;
6523 crate::appliance::tunnel_address_answer(self.stores.root(), address, window_open)
6524 }
6525
6526 #[cfg(feature = "wg-client")]
6527 fn set_tunnel_address(&self, address: &str) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
6528 crate::appliance::tunnel_address_answer(self.stores.root(), address, true)
6532 }
6533
6534 fn twin_csr(&self) -> ControlResult<String> {
6535 crate::twin_tls::request(self.stores.root()).map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6536 }
6537
6538 fn install_twin_certificate(&self, cert_pem: &str, ca_pem: &str, replication_port: u16, replication_host: &str) -> ControlResult<i64> {
6539 let host = (!replication_host.is_empty()).then_some(replication_host);
6540 crate::twin_tls::install_at(self.stores.root(), cert_pem, ca_pem, replication_port, host)
6541 .map(|expiry| expiry.days_left)
6542 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6543 }
6544
6545 fn issue_twin_certificate(&self, principal: &str, csr_pem: &str) -> ControlResult<(String, String, u16, String)> {
6546 crate::twin_tls::issue_for_twin(&self.stores, principal, csr_pem)
6547 .map(|(cert, ca, port)| (cert, ca, port, crate::twin_tls::replication_host(self.stores.root()).unwrap_or_default()))
6548 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6549 }
6550
6551 fn twin_git_listener_needs_restart(&self) -> bool {
6552 crate::twin::git_listener_needs_restart(&self.stores)
6553 }
6554
6555 fn restore_begin(&self, owner: &str) -> ControlResult<u64> {
6556 crate::twin::restore_begin_by(&self.stores, owner).map_err(|err| {
6557 if err.downcast_ref::<crate::twin::RestoreNotYours>().is_some() {
6559 ControlError::PermissionDenied(format!("{err:#}"))
6560 } else {
6561 ControlError::FailedPrecondition(format!("{err:#}"))
6562 }
6563 })
6564 }
6565
6566 fn restore_keys(&self) -> ControlResult<Vec<String>> {
6570 #[allow(unused_mut)]
6571 let mut keys = self.restore_keys.clone();
6572 #[cfg(feature = "wg-client")]
6573 keys.extend(crate::fed_principals::restore_keys_at(self.stores.root()).map_err(ControlError::Unavailable)?);
6574 keys.sort();
6575 keys.dedup();
6576 Ok(keys)
6577 }
6578
6579 fn restore_room(&self, announced_bytes: u64) -> ControlResult<()> {
6580 crate::twin::room_for_restore(self.stores.root(), announced_bytes).map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6581 }
6582
6583 fn restore_intake<'a>(&'a self, owner: &str) -> ControlResult<Option<Box<dyn gunnar_grpc::RestoreIntake + 'a>>> {
6587 match crate::twin::intake::Intake::open(&self.stores, owner) {
6588 Ok(Some(intake)) => Ok(Some(Box::new(intake))),
6589 Ok(None) => Ok(None),
6590 Err(err) if err.downcast_ref::<crate::twin::RestoreNotYours>().is_some() => Err(ControlError::PermissionDenied(format!("{err:#}"))),
6592 Err(err) => Err(ControlError::FailedPrecondition(format!("{err:#}"))),
6593 }
6594 }
6595
6596 fn restore_apply(&self, entry: gunnar_proto::LogEntry) -> ControlResult<bool> {
6597 let e = gunnar_core::txlog::Entry { seq: entry.seq, kind: entry.kind as u16, meta: entry.meta, blob: entry.blob };
6598 match crate::twin::restore_apply(&self.stores, &e) {
6599 Ok(crate::twin::Applied::Lost(_)) => Ok(true),
6600 Ok(_) => Ok(false),
6601 Err(err) => Err(ControlError::Internal(format!("restoring seq {}: {err:#}", e.seq))),
6602 }
6603 }
6604
6605 fn restore_finish(&self, announced_head: u64, log_id_of_source: &str) -> ControlResult<u64> {
6606 let head = crate::twin::restore_finish_streamed(&self.stores, announced_head, log_id_of_source).map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
6607 crate::twin::restart_after_restore();
6609 Ok(head)
6610 }
6611
6612 fn snapshot_stream(&self, sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>) -> ControlResult<u64> {
6613 self.snapshot_stream_from(0, 0, "", sink)
6614 }
6615
6616 fn snapshot_stream_from(&self, from_item: u64, from_offset: u64, expect_id: &str, sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>) -> ControlResult<u64> {
6621 let mut pieces = |p: crate::twin::snapshot::Piece<'_>, pos: Option<crate::twin::snapshot::PiecePos<'_>>| -> anyhow::Result<()> {
6623 sink(gunnar_proto::LogChunk {
6624 seq: 0,
6625 kind: u32::from(p.kind),
6626 meta: p.meta.to_vec(),
6627 blob_len: p.blob_len,
6628 blob: p.blob.to_vec(),
6629 last: p.last,
6630 restore_head: 0,
6631 restore_bytes: 0,
6632 log_id: String::new(),
6633 offset: pos.map(|q| q.offset),
6634 entry_crc: pos.and_then(|q| q.crc),
6635 snapshot_item: pos.map(|q| q.item),
6636 snapshot_id: pos.map(|q| q.id.to_owned()).unwrap_or_default(),
6637 })
6638 .map_err(|err| anyhow::anyhow!("{err}"))
6639 };
6640 crate::twin::snapshot::serve_from(&self.stores, from_item, from_offset, expect_id, &mut pieces).map_err(|err| match err.downcast_ref::<crate::twin::snapshot::SnapshotChanged>() {
6641 Some(changed) => ControlError::FailedPrecondition(changed.to_string()),
6642 None => ControlError::Internal(format!("producing a snapshot: {err:#}")),
6643 })
6644 }
6645
6646 fn log_retain(&self, from: u64) -> ControlResult<gunnar_proto::LogHeadResponse> {
6647 let log = self.stores.txlog();
6648 if log.is_follower() {
6649 return Err(ControlError::FailedPrecondition("retention: a twin's log is the primary's numbers; retain on the primary".into()));
6650 }
6651 let from = if from == 0 { log.head_seq() + 1 } else { from };
6652 let first = log.retain_from(from).map_err(|err| ControlError::Unavailable(format!("retaining the transaction log from {from}: {err}")))?;
6653 tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, from, first, head = log.head_seq(), "the transaction log now retains from {first}; a twin behind it takes a snapshot");
6654 self.log_head()
6655 }
6656
6657 fn log_layout(&self) -> ControlResult<Vec<(u64, u64, u64)>> {
6658 let log = self.stores.txlog();
6659 if log.is_follower() {
6660 return Err(ControlError::FailedPrecondition("retention: a twin's log is the primary's numbers; retain on the primary".into()));
6661 }
6662 Ok(log.layout())
6663 }
6664
6665 fn data_volume(&self) -> Option<(u64, u64)> {
6666 if self.stores.txlog().is_released() {
6669 return None;
6670 }
6671 let vfs = rustix::fs::statvfs(self.stores.root()).ok()?;
6672 let total = vfs.f_blocks.saturating_mul(vfs.f_frsize);
6673 let free = vfs.f_bfree.saturating_mul(vfs.f_frsize);
6674 (total > 0).then(|| (total, total.saturating_sub(free)))
6675 }
6676
6677 fn twin_owed_read(&self) -> ControlResult<Vec<u8>> {
6678 match std::fs::read(self.stores.root().join(crate::twin::TWIN_OWED_FILE)) {
6679 Ok(b) => Ok(b),
6680 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
6681 Err(e) => Err(ControlError::Unavailable(format!("reading {}: {e}", crate::twin::TWIN_OWED_FILE))),
6682 }
6683 }
6684
6685 fn twin_owed_write(&self, bytes: &[u8]) -> ControlResult<()> {
6686 if self.stores.txlog().is_released() {
6687 return Err(ControlError::Unavailable("the data set is released for a growth; twin-owed.json is written after the remount".into()));
6688 }
6689 crate::durable::write_atomic_unframed(self.stores.root(), crate::twin::TWIN_OWED_FILE, bytes).map_err(|e| ControlError::Unavailable(format!("writing {}: {e:#}", crate::twin::TWIN_OWED_FILE)))
6691 }
6692
6693 fn restore_snapshot_chunk(&self, chunk: &gunnar_proto::LogChunk) -> ControlResult<Option<u64>> {
6694 let kind = u16::try_from(chunk.kind).map_err(|_| ControlError::InvalidArgument(format!("snapshot piece of kind {}", chunk.kind)))?;
6695 let piece = crate::twin::snapshot::Piece { kind, meta: &chunk.meta, blob_len: chunk.blob_len, blob: &chunk.blob, last: chunk.last };
6696 crate::twin::snapshot::restore_piece(&self.stores, piece).map_err(|err| ControlError::Internal(format!("installing the snapshot: {err:#}")))
6697 }
6698
6699 fn health_alarms(&self) -> Vec<gunnar_proto::HealthAlarm> {
6700 #[allow(unused_mut)]
6701 let mut alarms = Vec::new();
6702 if !self.stores.is_twin() && crate::twin::is_restoring(self.stores.root()) {
6704 alarms.push(gunnar_proto::HealthAlarm {
6705 event: "twin.restoring_at_open".into(),
6706 count: 1,
6707 since_unix_ms: 0,
6708 detail: "this box is MID-RESTORE: it originates no transactions and refuses every \
6709 control-plane write until a restore finishes — resend the twin's log"
6710 .into(),
6711 });
6712 }
6713 if !self.stores.is_twin() {
6716 if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
6717 alarms.push(gunnar_proto::HealthAlarm { event: "twin.fenced_by_generation".into(), count: 1, since_unix_ms: 0, detail: why });
6718 }
6719 }
6720 if self.stores.is_twin() {
6723 if let Some(d) = crate::twin::snapshot::dropped(self.stores.root()) {
6724 let too_small = crate::twin::snapshot::too_small_total(self.stores.root())
6725 .map(|t| format!(" The resync did not fit even then: this twin's volume ({t} bytes) must be grown to at least main's data set before it is tried again ({}).", gunnar_grpc::TWIN_TOO_SMALL))
6726 .unwrap_or_default();
6727 alarms.push(gunnar_proto::HealthAlarm {
6728 event: "twin.resync_without_copy".into(),
6729 count: 1,
6730 since_unix_ms: d.at_unix_ms,
6731 detail: format!("{}.{too_small}", d.sentence()),
6732 });
6733 }
6734 }
6735 if !self.stores.is_twin() {
6738 let (total, _) = crate::twin::volume_bytes(self.stores.root());
6739 match crate::twin::txlog_alarm_cap(std::env::var(crate::twin::TXLOG_ALARM_ENV).ok().as_deref(), total) {
6740 Ok(Some((cap, how))) => alarms.extend(crate::twin::txlog_size_alarm(self.stores.txlog().bytes(), cap, &how)),
6741 Ok(None) => {}
6742 Err(why) => alarms.push(gunnar_proto::HealthAlarm { event: crate::twin::TXLOG_SIZE_ALARM.into(), count: 1, since_unix_ms: 0, detail: why }),
6743 }
6744 }
6745 #[cfg(feature = "metered")]
6746 if let Some(a) = self.entitlement.as_ref().and_then(|slot| slot.alarm()) {
6747 alarms.push(a);
6748 }
6749 if let Some(a) = self.grow.as_ref().and_then(|g| g.alarm()) {
6752 alarms.push(a);
6753 }
6754 alarms.extend(crate::feed::alarms());
6757 alarms
6758 }
6759
6760 fn is_twin(&self) -> bool {
6761 self.stores.is_twin()
6762 }
6763
6764 fn log_follows(&self) -> bool {
6765 self.stores.is_twin() || self.stores.txlog().is_follower()
6766 }
6767
6768 fn topology_facts(&self) -> gunnar_grpc::topology::Facts {
6769 crate::topology::facts(self.stores.root())
6770 }
6771
6772 fn maintenance_reasons(&self) -> Vec<String> {
6773 #[allow(unused_mut)]
6774 let mut why = Vec::new();
6775 #[cfg(feature = "metered")]
6776 if self.entitlement.as_ref().is_some_and(|slot| slot.maintenance()) {
6777 why.push(crate::entitlement::KEY_MISMATCH_ALARM.to_string());
6778 }
6779 why
6780 }
6781
6782 fn primary_heartbeat(&self, silent_after_ms: u64) -> Option<gunnar_proto::PrimaryHeartbeat> {
6783 let cfg = crate::twin::TwinConfig::load(self.stores.root()).ok().flatten()?;
6784 let st = crate::twin::status(self.stores.root());
6788 let (last, primary_head, applied) = match &st {
6789 Some(s) => (s.last_answer_unix_ms, s.primary_head_seq, s.applied_seq),
6790 None => (0, 0, self.stores.txlog().head_seq()),
6791 };
6792 let own = st.as_ref().map(|s| (s.state.to_string(), s.why.clone()));
6793 Some(gunnar_grpc::heartbeat::primary_view(
6794 &cfg.primary,
6795 last,
6796 primary_head,
6797 applied,
6798 crate::twin::now_ms(),
6799 silent_after_ms,
6800 own.as_ref().map(|(a, b)| (a.as_str(), b.as_str())),
6801 ))
6802 }
6803
6804 fn log_head(&self) -> ControlResult<gunnar_proto::LogHeadResponse> {
6805 self.log_head_signed("")
6806 }
6807
6808 fn log_head_signed(&self, nonce: &str) -> ControlResult<gunnar_proto::LogHeadResponse> {
6809 if !nonce.trim().is_empty() {
6820 let root = self.stores.root();
6821 if crate::twin::is_restoring(root) {
6822 return Err(ControlError::Unavailable(
6823 "this box is being restored from a twin and is not a primary yet: it answers a twin's poll \
6824 once the replay has finished and the box has restarted"
6825 .to_owned(),
6826 ));
6827 }
6828 if crate::twin::reseal_pending(root) {
6829 return Err(ControlError::Unavailable(
6830 "this box finished a restore and must restart before it answers a twin: its host key is \
6831 sealed under the key-encrypting key the restart re-seals, so until then it cannot prove \
6832 which box it is"
6833 .to_owned(),
6834 ));
6835 }
6836 }
6837 let mut head = crate::twin::own_head(&self.stores);
6838 let root = self.stores.root();
6849 let generation = if crate::twin::is_twin_root(root) {
6850 crate::twin::pinned_generation(root)
6851 } else {
6852 crate::twin::log_generation(root)
6853 };
6854 head.generation = generation.map_err(|err| ControlError::Internal(format!("this box's log generation: {err:#}")))?;
6855 let (key, sig) =
6856 crate::twin::sign_head_answer(&self.stores, nonce, &head.log_id, head.first_seq, head.head_seq, head.generation);
6857 head.answer_key = key;
6858 head.answer_sig = sig;
6859 Ok(head)
6860 }
6861
6862 fn fence_by_generation(&self, proof: &gunnar_proto::GenerationProof) -> ControlResult<()> {
6863 crate::twin::fence_by_proof(&self.stores, proof)
6864 .map(drop)
6865 .map_err(|err| ControlError::Internal(format!("fencing this box off as an old generation of its log: {err:#}")))
6866 }
6867
6868 fn log_fetch(
6869 &self,
6870 from: u64,
6871 to: u64,
6872 max_bytes: u64,
6873 ) -> ControlResult<Vec<gunnar_proto::LogEntry>> {
6874 let log = self.readable_log()?;
6875 let entries = crate::twin::fetch_range(log, from, to, max_bytes).map_err(log_read_refusal)?;
6876 Ok(entries
6877 .into_iter()
6878 .map(|e| gunnar_proto::LogEntry {
6879 seq: e.seq,
6880 kind: e.kind as u32,
6881 meta: e.meta,
6882 blob: e.blob,
6883 })
6884 .collect())
6885 }
6886
6887 fn log_fetch_stream(&self, from: u64, to: u64, max_bytes: u64) -> ControlResult<gunnar_grpc::seam::LogFetchProducer> {
6892 self.log_fetch_stream_at(from, 0, to, max_bytes)
6893 }
6894
6895 fn log_fetch_stream_at(&self, from: u64, from_offset: u64, to: u64, max_bytes: u64) -> ControlResult<gunnar_grpc::seam::LogFetchProducer> {
6898 let log = self.readable_log()?;
6899 crate::twin::check_range(log, from, to).map_err(log_read_refusal)?;
6900 let stores = Arc::clone(&self.stores);
6901 Ok(Box::new(move |sink| {
6902 let mut out = ChunkOut { sink, open: None };
6903 crate::twin::stream_range_at(stores.txlog(), from, from_offset, to, max_bytes, &mut out).map(drop).map_err(|err| {
6904 let err = match err.downcast::<ControlError>() {
6905 Ok(said) => return said,
6906 Err(other) => other,
6907 };
6908 match err.downcast::<gunnar_core::txlog::Error>() {
6909 Ok(e) => log_read_refusal(e),
6910 Err(other) => ControlError::Internal(format!("transaction log: {other:#}")),
6911 }
6912 })
6913 }))
6914 }
6915
6916 #[cfg(feature = "replication")]
6917 fn replication_status(
6918 &self,
6919 include_refs: bool,
6920 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
6921 self.replication_slot()?;
6922 self.replication_snapshot(include_refs)
6923 }
6924
6925 #[cfg(feature = "replication")]
6937 fn replication_flush(
6938 &self,
6939 request: &gunnar_proto::FlushReplicationRequest,
6940 ) -> ControlResult<gunnar_proto::FlushReplicationResponse> {
6941 let slot = self.replication_slot()?;
6942 let began = gunnar_grpc::bus::now_unix_ms();
6943 let started = std::time::Instant::now();
6944 let ceiling = if request.timeout_secs == 0 { abdicate_flush_secs() } else { u64::from(request.timeout_secs) };
6945 let Some(replicator) = slot.replicator() else {
6946 return Err(ControlError::FailedPrecondition(
6947 "no replicator is running on this box (no standby configured, or this box is not a primary); there is nothing to flush and nothing to prove".to_owned(),
6948 ));
6949 };
6950 if request.resync {
6951 replicator
6952 .resync_scoped(None, None, &[])
6953 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6954 }
6955 let deadline = started + std::time::Duration::from_secs(ceiling);
6956 loop {
6957 let drained = replicator.status().iter().all(|s| s.pending == 0);
6958 if drained || std::time::Instant::now() >= deadline {
6959 break;
6960 }
6961 std::thread::sleep(std::time::Duration::from_millis(100));
6962 }
6963 let status = replicator.status();
6964 let counters = replicator.counters();
6965 let pending: u64 = status.iter().map(|s| s.pending as u64).sum();
6966 let needs_full_resync = status.iter().any(|s| s.needs_full_resync);
6967 let last_success = status
6968 .iter()
6969 .map(|s| s.last_success_unix_ms.unwrap_or(0))
6970 .min()
6971 .unwrap_or(0);
6972 let in_flight = self.admission.as_ref().map(|a| a.in_flight()).unwrap_or(0);
6973 let catalog_repos = self
6974 .stores
6975 .catalog()
6976 .counts()
6977 .map(|(_, repos)| repos as u64)
6978 .unwrap_or(0);
6979 let standbys = self.replication_snapshot(false)?.standbys;
6980 let caught_up = in_flight == 0
6981 && !status.is_empty()
6982 && pending == 0
6983 && !needs_full_resync
6984 && status.iter().all(|s| s.last_success_unix_ms.is_some_and(|t| t >= began) || s.pending == 0 && s.last_success_unix_ms.is_some());
6985 let verdict = if status.is_empty() {
6986 "no standby is configured: nothing was flushed and nothing is proven".to_owned()
6987 } else if caught_up {
6988 format!(
6989 "CAUGHT UP as far as replication can prove: {} standby(s), nothing pending, no full-resync hole, {} transfer(s) in flight, {} bus gap(s) since boot{}. What this does NOT prove (DATA-SET-GROWTH-FLOW.md §2.1): refs outside heads/tags, LFS objects, the control database, visibility, tombstones, and — unless `resync` ran once since the twin existed — repositories nobody pushed since",
6990 status.len(),
6991 in_flight,
6992 counters.bus_gaps,
6993 if request.resync { " (every catalog repository was enqueued)" } else { "" }
6994 )
6995 } else {
6996 let names: Vec<String> = status
6997 .iter()
6998 .filter(|s| s.pending > 0 || s.needs_full_resync)
6999 .map(|s| format!("{} (pending {}{})", s.name, s.pending, if s.needs_full_resync { ", needs full resync" } else { "" }))
7000 .collect();
7001 format!(
7002 "NOT caught up after {} s: {} transfer(s) in flight, {} repository(ies) pending [{}]{}; the growth must not unmount the set on this",
7003 started.elapsed().as_secs(),
7004 in_flight,
7005 pending,
7006 names.join(", "),
7007 if needs_full_resync { " — a standby has a hole retrying will not close: run `replication resync`" } else { "" }
7008 )
7009 };
7010 tracing::info!(
7011 target: "gunnar::replication",
7012 event = "replication.flush",
7013 caught_up,
7014 in_flight,
7015 pending,
7016 bus_gaps = counters.bus_gaps,
7017 needs_full_resync,
7018 waited_ms = started.elapsed().as_millis() as u64,
7019 "{verdict}"
7020 );
7021 Ok(gunnar_proto::FlushReplicationResponse {
7022 caught_up,
7023 in_flight,
7024 pending,
7025 bus_gaps: counters.bus_gaps,
7026 needs_full_resync,
7027 last_success_unix_ms: last_success,
7028 flushed_at_unix_ms: began,
7029 waited_ms: started.elapsed().as_millis() as u64,
7030 standbys,
7031 catalog_repos,
7032 verdict,
7033 })
7034 }
7035
7036 #[cfg(feature = "replication")]
7050 fn replication_demote(&self) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7051 let slot = self.replication_slot()?;
7052 let mut config = slot
7053 .config()
7054 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7055 config.clear_targets();
7056 slot.apply(config)
7057 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7058 self.replication_snapshot(false)
7059 }
7060
7061 #[cfg(feature = "replication")]
7071 fn replication_set_role(
7072 &self,
7073 role: gunnar_proto::ReplicationRole,
7074 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7075 let slot = self.replication_slot()?;
7076 let mut config = slot
7077 .config()
7078 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7079 let wanted = crate::mirror::Role::from_proto(role);
7080 if config.role.is_browse() && wanted == crate::mirror::Role::Primary {
7088 return Err(ControlError::FailedPrecondition(
7089 "this box is a browse replica; it cannot be promoted to primary. It holds only \
7090 public repositories and serves anonymous reads — first `gunnar control \
7091 replication demote` it to a plain standby, then promote that"
7092 .into(),
7093 ));
7094 }
7095 config.role = wanted;
7096 slot.apply(config)
7097 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7098 self.replication_snapshot(false)
7099 }
7100
7101 #[cfg(feature = "replication")]
7123 fn replication_abdicate(&self) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7124 let slot = self.replication_slot()?;
7125
7126 let mut config = slot
7129 .config()
7130 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7131 config.role = crate::mirror::Role::Standby;
7132 slot.apply(config)
7133 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7134
7135 if let Some(replicator) = slot.replicator() {
7140 replicator
7141 .resync_scoped(None, None, &[])
7142 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7143 let deadline =
7144 std::time::Instant::now() + std::time::Duration::from_secs(abdicate_flush_secs());
7145 loop {
7146 let drained = replicator.status().iter().all(|s| s.pending == 0);
7147 if drained || std::time::Instant::now() >= deadline {
7148 break;
7149 }
7150 std::thread::sleep(std::time::Duration::from_millis(100));
7151 }
7152 }
7153
7154 let mut config = slot
7158 .config()
7159 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7160 config.clear_targets();
7161 slot.apply(config)
7162 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7163 self.replication_snapshot(false)
7164 }
7165
7166 #[cfg(feature = "replication")]
7180 fn replication_resync(
7181 &self,
7182 name: &str,
7183 repo: Option<&gunnar_proto::RepoPath>,
7184 except: &[gunnar_proto::RepoPath],
7185 ) -> ControlResult<gunnar_proto::ResyncReplicationResponse> {
7186 let slot = self.replication_slot()?;
7187 let Some(replicator) = slot.replicator() else {
7188 return Err(ControlError::Unavailable(format!(
7189 "no replicator is running, so nothing would drain a backfill queue. \
7190 `replication.json` holds {} target(s); the server has not started them.",
7191 slot.config().map(|c| c.targets.len()).unwrap_or(0)
7192 )));
7193 };
7194 let only = match repo {
7195 None => None,
7196 Some(path) => Some(Self::repo_ref(path)?),
7197 };
7198 let except = except
7199 .iter()
7200 .map(Self::repo_ref)
7201 .collect::<ControlResult<Vec<_>>>()?;
7202 let reports = replicator
7203 .resync_scoped(Some(name), only.as_ref(), &except)
7204 .map_err(|err| ControlError::NotFound(format!("{err:#}")))?;
7205 Ok(gunnar_proto::ResyncReplicationResponse {
7206 standbys: reports
7207 .into_iter()
7208 .map(|r| gunnar_proto::ResyncStandby {
7209 standby: r.standby,
7210 in_scope: r.in_scope,
7211 queued: r.queued,
7212 already_pending: r.already_pending,
7213 refused: r.refused,
7214 sha1_repos: r.sha1_repos,
7215 excluded: r.excluded,
7216 })
7217 .collect(),
7218 status: Some(self.replication_snapshot(false)?),
7219 })
7220 }
7221
7222 #[cfg(feature = "webhooks")]
7225 fn webhook_add(
7226 &self,
7227 request: gunnar_proto::AddWebhookRequest,
7228 ) -> ControlResult<gunnar_proto::WebhookListResponse> {
7229 let slot = self.webhook_slot()?;
7230 let endpoint = crate::webhook::Endpoint {
7231 name: request.name,
7232 url: request.url,
7233 kinds: request.kinds,
7234 namespace: request.namespace,
7235 repo: request.repo,
7236 secret: Some(request.secret).filter(|s| !s.is_empty()),
7239 added_at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
7240 };
7241 slot.add(endpoint)
7244 .map_err(|err| ControlError::InvalidArgument(format!("{err:#}")))
7245 }
7246
7247 #[cfg(feature = "webhooks")]
7248 fn webhook_list(&self) -> ControlResult<gunnar_proto::WebhookListResponse> {
7249 self.webhook_slot()?
7250 .list_response()
7251 .map_err(|err| ControlError::Internal(format!("{err:#}")))
7252 }
7253
7254 #[cfg(feature = "webhooks")]
7255 fn webhook_remove(
7256 &self,
7257 name: &str,
7258 ) -> ControlResult<gunnar_proto::WebhookListResponse> {
7259 let slot = self.webhook_slot()?;
7260 slot.remove(name)
7261 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
7262 .ok_or_else(|| ControlError::NotFound(format!("no webhook endpoint named {name:?}")))
7266 }
7267}
7268
7269const MAX_RESPONSE_BYTES: u64 = 4 * 1024 * 1024;
7295
7296const READ_OBJECT_MEMORY_BUDGET: u64 = 512 * 1024 * 1024;
7305
7306const MAX_CONCURRENT_INFLATES: u64 = 8;
7325
7326const DEFAULT_MAX_OBJECT_BYTES: u64 = READ_OBJECT_MEMORY_BUDGET / MAX_CONCURRENT_INFLATES;
7352
7353impl CoreControlPlane {
7381 fn usage_readable(&self) -> ControlResult<()> {
7383 if self.stores.txlog().is_released() || crate::twin::maintenance::is_held(self.stores.root()) {
7384 return Err(ControlError::Unavailable(format!(
7385 "{} this box's data set is being grown (unmounted): its usage is read again after the remount — ask again",
7386 gunnar_grpc::PRIMARY_GROWING
7387 )));
7388 }
7389 Ok(())
7390 }
7391
7392 fn open_objects_of(&self, namespace: &str) -> u64 {
7394 let prefix = self.stores.repos_root().canonicalize().unwrap_or_else(|_| self.stores.repos_root().to_path_buf()).join(namespace);
7395 self.stores
7396 .engine()
7397 .open_object_counts()
7398 .into_iter()
7399 .filter(|(dir, _)| dir.parent() == Some(prefix.as_path()))
7400 .map(|(_, n)| n)
7401 .sum()
7402 }
7403
7404 fn usage_row(
7407 &self,
7408 name: &str,
7409 held: &crate::usage_ledger::Held,
7410 objects: u64,
7411 measured_at_unix_ms: i64,
7412 ) -> gunnar_proto::NamespaceUsage {
7413 let mut row = gunnar_proto::NamespaceUsage {
7414 namespace: name.to_owned(),
7415 measured_at_unix_ms,
7416 pack_bytes: held.bytes.pack,
7417 cache_bytes: held.bytes.cache,
7418 lfs_bytes: held.bytes.lfs,
7419 tombstoned_bytes: held.tombstoned_bytes,
7420 repos: held.repos,
7421 objects,
7422 ..Default::default()
7423 };
7424 fill_flow(&mut row, self.stores.flow(), name, measured_at_unix_ms);
7425 row
7426 }
7427
7428 fn usage_all(&self) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
7432 let catalog = self.stores.catalog();
7433 let namespaces: Vec<String> = catalog
7434 .namespaces()
7435 .map_err(internal)?
7436 .into_iter()
7437 .map(|ns| ns.name)
7438 .collect();
7439 let repos = catalog.list(None).map_err(internal)?;
7440 let tombstones = self.stores.tombstones().unwrap_or_default();
7441 let open_objects: std::collections::HashMap<std::path::PathBuf, u64> =
7442 self.stores.engine().open_object_counts().into_iter().collect();
7443 let flow = self.stores.flow();
7444 let measured_at_unix_ms = gunnar_core::now_ms();
7445 let mut by_ns: std::collections::HashMap<&str, Vec<&gunnar_core::Repo>> = std::collections::HashMap::new();
7446 for repo in &repos {
7447 by_ns.entry(repo.namespace.as_str()).or_default().push(repo);
7448 }
7449 let mut out = Vec::with_capacity(namespaces.len());
7450 for name in namespaces {
7451 let mut row = gunnar_proto::NamespaceUsage {
7452 namespace: name.clone(),
7453 measured_at_unix_ms,
7454 ..Default::default()
7455 };
7456 for repo in by_ns.get(name.as_str()).map(Vec::as_slice).unwrap_or(&[]) {
7457 let path = RepoRef::new(&repo.namespace, &repo.name).map_err(|err| {
7458 ControlError::Internal(format!("a catalogued repository has an illegal path: {err}"))
7459 })?;
7460 let dir = self
7461 .stores
7462 .repo_dir(&path)
7463 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7464 let bytes = repo_bytes(&dir);
7465 row.repos += 1;
7466 row.pack_bytes += bytes.pack;
7467 row.cache_bytes += bytes.cache;
7468 row.lfs_bytes += bytes.lfs;
7469 row.objects += open_objects.get(&dir).copied().unwrap_or(0);
7470 }
7471 row.tombstoned_bytes = tombstones
7472 .iter()
7473 .filter(|t| t.path.namespace() == name)
7474 .map(|t| t.bytes)
7475 .sum();
7476 fill_flow(&mut row, flow, &name, measured_at_unix_ms);
7477 out.push(row);
7478 }
7479 Ok(out)
7480 }
7481}
7482
7483fn fill_flow(row: &mut gunnar_proto::NamespaceUsage, flow: &gunnar_core::FlowLedger, name: &str, now_unix_ms: i64) {
7498 let glance = flow.glance(name, now_unix_ms);
7499 row.bytes_in_today = glance.today.bytes_in;
7500 row.bytes_out_today = glance.today.bytes_out;
7501 row.pushes_today = glance.today.pushes;
7502 row.fetches_today = glance.today.fetches;
7503 row.refused_today = glance.today.refused();
7504 row.last_push_unix_ms = glance.last_push_unix_ms;
7505 row.last_fetch_unix_ms = glance.last_fetch_unix_ms;
7506}
7507
7508fn object_dir_stats(git_dir: &std::path::Path) -> (u64, u64) {
7514 fn walk(dir: &std::path::Path, loose: &mut u64, bytes: &mut u64, in_loose_shard: bool) {
7515 let Ok(entries) = std::fs::read_dir(dir) else {
7516 return;
7517 };
7518 for entry in entries.flatten() {
7519 let Ok(kind) = entry.file_type() else {
7520 continue;
7521 };
7522 let name = entry.file_name();
7523 let name = name.to_string_lossy();
7524 if kind.is_dir() {
7525 let shard = name.len() == 2 && name.chars().all(|c| c.is_ascii_hexdigit());
7527 walk(&entry.path(), loose, bytes, shard);
7528 } else if kind.is_file() {
7529 if in_loose_shard {
7530 *loose += 1;
7531 }
7532 if let Ok(meta) = entry.metadata() {
7533 *bytes += meta.len();
7534 }
7535 }
7536 }
7537 }
7538 let mut loose = 0;
7539 let mut bytes = 0;
7540 walk(&git_dir.join("objects"), &mut loose, &mut bytes, false);
7541 (loose, bytes)
7542}
7543
7544#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
7564pub struct RepoBytes {
7565 pub pack: u64,
7567 pub cache: u64,
7569 pub lfs: u64,
7571}
7572
7573impl RepoBytes {
7574 pub fn hot(&self) -> u64 {
7576 self.pack + self.cache
7577 }
7578}
7579
7580pub fn repo_bytes(git_dir: &std::path::Path) -> RepoBytes {
7582 fn walk(dir: &std::path::Path, acc: &mut RepoBytes) {
7583 let Ok(entries) = std::fs::read_dir(dir) else {
7584 return;
7585 };
7586 for entry in entries.flatten() {
7587 let Ok(meta) = entry.metadata() else {
7588 continue;
7589 };
7590 if meta.is_dir() {
7591 walk(&entry.path(), acc);
7592 } else if entry
7593 .file_name()
7594 .to_string_lossy()
7595 .starts_with("objects.exploded")
7596 {
7597 acc.cache += meta.len();
7598 } else {
7599 acc.pack += meta.len();
7600 }
7601 }
7602 }
7603 let mut acc = RepoBytes::default();
7604 let Ok(entries) = std::fs::read_dir(git_dir) else {
7605 return acc;
7606 };
7607 for entry in entries.flatten() {
7608 let Ok(meta) = entry.metadata() else {
7609 continue;
7610 };
7611 if meta.is_dir() && entry.file_name() == gunnar_lfs::fs_store::LFS_DIR {
7612 acc.lfs += crate::stores::dir_bytes(&entry.path());
7613 } else if meta.is_dir() {
7614 walk(&entry.path(), &mut acc);
7615 } else {
7616 acc.pack += meta.len();
7617 }
7618 }
7619 acc
7620}
7621
7622impl gunnar_grpc::Identity for CoreControlPlane {
7644 fn awaiting_enrolment(&self, fingerprint: &str) -> Option<String> {
7647 self.stores.awaiting_enrolment(fingerprint)
7648 }
7649
7650 fn registered_key(
7651 &self,
7652 fingerprint: &str,
7653 ) -> ControlResult<Option<gunnar_grpc::RegisteredKey>> {
7654 let Ok(parsed) = gunnar_core::Fingerprint::parse(fingerprint) else {
7658 return Ok(None);
7659 };
7660 let Some(principal) = self
7663 .stores
7664 .auth()
7665 .principal_by_fingerprint(&parsed)
7666 .map_err(internal)?
7667 else {
7668 return Ok(None);
7669 };
7670 let key = self
7671 .stores
7672 .auth()
7673 .keys_of(&principal.id)
7674 .map_err(internal)?
7675 .into_iter()
7676 .find(|key| key.fingerprint_sha256 == fingerprint);
7677 let machine = principal.kind == gunnar_core::PrincipalKind::Machine;
7678 Ok(key.map(|key| gunnar_grpc::RegisteredKey {
7679 principal_id: principal.id,
7680 machine,
7681 openssh: key.openssh,
7685 }))
7686 }
7687
7688 fn registered_certificate(
7703 &self,
7704 der: &[u8],
7705 ) -> ControlResult<Option<gunnar_grpc::CertificateIdentity>> {
7706 let fingerprint = gunnar_core::auth::fingerprint_of_certificate(der);
7710 let Some(principal) = self
7711 .stores
7712 .auth()
7713 .principal_by_fingerprint(&fingerprint)
7714 .map_err(internal)?
7715 else {
7716 return Ok(None);
7717 };
7718 let registered = self
7724 .stores
7725 .auth()
7726 .keys_of(&principal.id)
7727 .map_err(internal)?
7728 .into_iter()
7729 .any(|key| key.fingerprint_sha256 == fingerprint.to_string());
7730 if !registered {
7731 return Ok(None);
7732 }
7733 Ok(Some(gunnar_grpc::CertificateIdentity {
7734 principal_id: principal.id,
7735 fingerprint: fingerprint.to_string(),
7736 }))
7737 }
7738
7739 fn narrowed_methods(&self, principal_id: &str) -> ControlResult<Option<Vec<String>>> {
7742 if principal_id == gunnar_grpc::REPAIR_PRINCIPAL {
7745 return Ok(Some(gunnar_grpc::REPAIR_METHODS.iter().map(|m| m.to_string()).collect()));
7746 }
7747 #[cfg(feature = "wg-client")]
7750 return crate::fed_principals::narrowed_of(self.stores.root(), principal_id).map_err(ControlError::Internal);
7751 #[cfg(not(feature = "wg-client"))]
7752 {
7753 let _ = principal_id;
7754 Ok(None)
7755 }
7756 }
7757
7758 fn may(
7759 &self,
7760 principal_id: &str,
7761 access: &gunnar_grpc::Access,
7762 level: gunnar_proto::GrantLevel,
7763 ) -> ControlResult<bool> {
7764 let level = level_in(level)?;
7765 let auth = self.stores.auth();
7766 match access {
7767 gunnar_grpc::Access::Server => auth.may_server(principal_id, level).map_err(internal),
7768 gunnar_grpc::Access::Namespace(namespace) => auth
7769 .may_namespace(principal_id, namespace, level)
7770 .map_err(internal),
7771 gunnar_grpc::Access::Repo(path) => {
7772 let repo_ref = Self::repo_ref(path)?;
7773 let Some(repo) = self.stores.catalog().get(&repo_ref).map_err(internal)? else {
7777 return Ok(false);
7778 };
7779 auth.may(principal_id, &repo, level).map_err(internal)
7780 }
7781 }
7782 }
7783
7784 fn may_each_repo(
7787 &self,
7788 principal_id: &str,
7789 paths: &[RepoPath],
7790 level: gunnar_proto::GrantLevel,
7791 ) -> ControlResult<Vec<bool>> {
7792 let level = level_in(level)?;
7793 let refs = paths.iter().map(Self::repo_ref).collect::<ControlResult<Vec<_>>>()?;
7794 let rows = self.stores.catalog().get_many(&refs).map_err(internal)?;
7795 let known: Vec<gunnar_core::model::Repo> = rows.iter().flatten().cloned().collect();
7796 let mut answers = self.stores.auth().may_each(principal_id, &known, level).map_err(internal)?.into_iter();
7797 Ok(rows.iter().map(|row| row.is_some() && answers.next().unwrap_or(false)).collect())
7798 }
7799}
7800
7801fn protection_out(rule: &gunnar_policy::ProtectionRule) -> BranchProtection {
7813 BranchProtection {
7814 ref_pattern: rule.pattern.as_str().to_owned(),
7815 allow_force_push: rule.allow_force_push,
7816 allow_delete: rule.allow_delete,
7817 require_signature: rule.require_signature,
7818 allowed_principal_ids: rule.allowed_principals.clone().unwrap_or_default(),
7821 max_object_bytes: 0,
7822 max_objects_per_push: 0,
7823 forbid_self_approval: rule.forbid_self_approval,
7824 required_approvals: rule.required_approvals,
7825 }
7826}
7827
7828fn ref_name_policy_out(policy: &gunnar_policy::RefNamePolicy) -> RefNamePolicy {
7829 RefNamePolicy {
7830 allow_patterns: policy.allow.iter().map(|p| p.as_str().to_owned()).collect(),
7831 deny_patterns: policy.deny.iter().map(|p| p.as_str().to_owned()).collect(),
7832 }
7833}
7834
7835fn mirror_out(record: &gunnar_core::MirrorRecord) -> MirrorConfig {
7844 let (namespace, repo) = record
7845 .repo
7846 .split_once('/')
7847 .unwrap_or((record.repo.as_str(), ""));
7848 MirrorConfig {
7849 repo: Some(RepoPath::new(namespace, repo)),
7850 remote_url: record.remote_url.clone(),
7851 prune: record.prune,
7852 state: if record.last_error.is_empty() {
7853 gunnar_proto::MirrorState::Idle as i32
7854 } else {
7855 gunnar_proto::MirrorState::Failed as i32
7856 },
7857 added_at_unix_ms: record.added_at_unix_ms,
7858 last_sync_at_unix_ms: record.last_sync_at_unix_ms,
7859 last_error: record.last_error.clone(),
7860 sync_count: record.sync_count,
7861 credential: credential_kind_out(record.credential) as i32,
7865 credential_username: record.credential_username.clone(),
7866 job: None,
7869 repo_pending: record.creates_repo,
7870 lfs_not_copied: record.lfs_detected,
7871 }
7872}
7873
7874fn credential_kind_in(kind: gunnar_proto::MirrorCredentialKind) -> gunnar_core::MirrorCredentialKind {
7876 match kind {
7877 gunnar_proto::MirrorCredentialKind::Unspecified => gunnar_core::MirrorCredentialKind::None,
7878 gunnar_proto::MirrorCredentialKind::SshKey => gunnar_core::MirrorCredentialKind::SshKey,
7879 gunnar_proto::MirrorCredentialKind::Token => gunnar_core::MirrorCredentialKind::Token,
7880 }
7881}
7882
7883fn credential_kind_out(kind: gunnar_core::MirrorCredentialKind) -> gunnar_proto::MirrorCredentialKind {
7885 match kind {
7886 gunnar_core::MirrorCredentialKind::None => gunnar_proto::MirrorCredentialKind::Unspecified,
7887 gunnar_core::MirrorCredentialKind::SshKey => gunnar_proto::MirrorCredentialKind::SshKey,
7888 gunnar_core::MirrorCredentialKind::Token => gunnar_proto::MirrorCredentialKind::Token,
7889 }
7890}
7891
7892fn read_credential(
7903 stores: &Arc<ServerStores>,
7904 repo: &gunnar_core::RepoRef,
7905) -> Result<crate::mirror_fetch::Credential, String> {
7906 let record = stores
7907 .mirrors()
7908 .get(repo)
7909 .map_err(|err| format!("reading the mirror registration for {}: {err:#}", repo.key()))?
7910 .ok_or_else(|| format!("no mirror is registered for {} any more", repo.key()))?;
7911 if !record.credential.is_set() {
7912 return Ok(crate::mirror_fetch::Credential::None);
7913 }
7914 let secrets = stores
7915 .secrets()
7916 .map_err(|err| format!("the sealed secret store is not available: {err:#}"))?;
7917 let key = gunnar_core::credential_secret_key(repo);
7918 let sealed = gunnar_store::SecretStore::get(&secrets, &key)
7919 .map_err(|err| format!("the mirror credential for {} could not be read: {err}", repo.key()))?
7923 .ok_or_else(|| {
7924 format!(
7925 "{} is registered as having {}, and the sealed value is not in the secret \
7926 store. Set it again with `Mirror.SetCredential`; dialling anonymously \
7927 instead would be refused by the upstream and read as a bad key.",
7928 repo.key(),
7929 record.credential.name()
7930 )
7931 })?;
7932 Ok(match record.credential {
7933 gunnar_core::MirrorCredentialKind::SshKey => {
7934 crate::mirror_fetch::Credential::SshKey(sealed)
7935 }
7936 gunnar_core::MirrorCredentialKind::Token => crate::mirror_fetch::Credential::Token {
7937 username: record.credential_username.clone(),
7938 secret: String::from_utf8(sealed).map_err(|_| {
7939 format!(
7940 "the token stored for {} is not utf-8, so it cannot be an HTTP header value",
7941 repo.key()
7942 )
7943 })?,
7944 },
7945 gunnar_core::MirrorCredentialKind::None => crate::mirror_fetch::Credential::None,
7946 })
7947}
7948
7949fn audit_line(stores: &ServerStores, repo: &gunnar_core::RepoRef, kind: &str, detail: &str, refused: bool) {
7953 tracing::info!(
7954 target: gunnar_grpc::AUDIT_TARGET,
7955 event = kind,
7956 repo = %repo.key(),
7957 detail,
7958 "mirror"
7959 );
7960 let Ok(Some(account)) = stores.auth().account_of_namespace(repo.namespace()) else { return };
7961 let event = gunnar_core::account_audit::AccountEvent {
7962 kind,
7963 actor: "",
7964 actor_role: "system",
7965 target: &repo.key(),
7966 detail,
7967 outcome: if refused {
7968 gunnar_core::account_audit::Outcome::Refused
7969 } else {
7970 gunnar_core::account_audit::Outcome::Ok
7971 },
7972 };
7973 if let Err(err) = stores.account_audit().append(&account, event, gunnar_core::now_ms()) {
7974 tracing::warn!(target: "gunnar::git", repo = %repo.key(), %err, "a mirror sync's audit line was not written");
7975 }
7976}
7977
7978fn audit_mirror_outcome(
7982 stores: &ServerStores,
7983 repo: &gunnar_core::RepoRef,
7984 order: &crate::mirror_queue::MirrorOrder,
7985 first: bool,
7986 error: Option<&str>,
7987 report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>,
7988) {
7989 let upstream = crate::mirror_fetch::scrub(&order.upstream);
7990 match (first, error) {
7991 (true, None) => {
7992 let refs = report.snapshot().map(|job| job.report.updates.len()).unwrap_or_default();
7993 audit_line(stores, repo, "mirror.import.done", &format!("imported {refs} reference(s) from {upstream}"), false);
7994 }
7995 (true, Some(why)) => audit_line(stores, repo, "mirror.import.failed", &format!("from {upstream}: {why}"), true),
7996 (false, Some(why)) => audit_line(stores, repo, "mirror.sync.failed", &format!("from {upstream}: {why}"), true),
7997 (false, None) => {}
7998 }
7999}
8000
8001fn mirror_job_out(
8011 path: &RepoPath,
8012 job: crate::mirror_queue::MirrorJob,
8013) -> gunnar_proto::MirrorJob {
8014 let report = job.report;
8018 let report_refused = report.refused;
8019 let (progress_done, progress_total) = job.progress.unwrap_or((0, 0));
8020 gunnar_proto::MirrorJob {
8021 job_id: job.job_id,
8022 path: Some(path.clone()),
8023 state: match job.state {
8024 crate::mirror_queue::MirrorState::Queued => {
8025 gunnar_proto::MirrorJobState::Queued as i32
8026 }
8027 crate::mirror_queue::MirrorState::Running => {
8028 gunnar_proto::MirrorJobState::Running as i32
8029 }
8030 crate::mirror_queue::MirrorState::Done => gunnar_proto::MirrorJobState::Done as i32,
8031 crate::mirror_queue::MirrorState::Failed => {
8032 gunnar_proto::MirrorJobState::Failed as i32
8033 }
8034 },
8035 queue_position: job.queue_position,
8036 has_progress: job.progress.is_some(),
8040 progress_done,
8041 progress_total,
8042 objects: report.objects,
8043 bytes: report.bytes,
8044 updates: report
8045 .updates
8046 .into_iter()
8047 .map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
8048 ref_name,
8049 old_oid,
8050 new_oid,
8051 })
8052 .collect(),
8053 error: job.error.unwrap_or_default(),
8054 enqueued_at_unix_ms: job.enqueued_at_unix_ms,
8055 started_at_unix_ms: job.started_at_unix_ms,
8056 finished_at_unix_ms: job.finished_at_unix_ms,
8057 refused_reason: match (job.state, report_refused) {
8060 (crate::mirror_queue::MirrorState::Failed, crate::mirror_queue::Refusal::Plan) => {
8061 gunnar_proto::MirrorRefusal::Plan as i32
8062 }
8063 (crate::mirror_queue::MirrorState::Failed, crate::mirror_queue::Refusal::PackCap) => {
8064 gunnar_proto::MirrorRefusal::PackCap as i32
8065 }
8066 _ => gunnar_proto::MirrorRefusal::Unspecified as i32,
8067 },
8068 }
8069}
8070
8071fn to_proto_membership(
8082 signed: gunnar_core::SignedMembershipEvent,
8083) -> gunnar_proto::SignedMembershipEvent {
8084 gunnar_proto::SignedMembershipEvent {
8085 event: Some(gunnar_proto::MembershipEvent {
8086 group: signed.event.group,
8087 member: signed.event.member,
8088 action: proto_action(signed.event.action) as i32,
8089 actor: signed.event.actor,
8090 at_unix_ms: signed.event.at_unix_ms,
8091 }),
8092 key_fingerprint: signed.key_fingerprint,
8093 signature: signed.signature,
8094 }
8095}
8096
8097fn from_proto_attestation(
8100 wire: gunnar_proto::SignedKeyAttestation,
8101) -> ControlResult<gunnar_core::SignedKeyAttestation> {
8102 let attestation = wire
8103 .attestation
8104 .ok_or_else(|| ControlError::InvalidArgument("`attestation.attestation` is required".to_owned()))?;
8105 Ok(gunnar_core::SignedKeyAttestation {
8106 attestation: gunnar_core::KeyAttestation {
8107 principal_id: attestation.principal_id,
8108 fingerprint: attestation.fingerprint,
8109 at_unix_ms: attestation.at_unix_ms,
8110 },
8111 key_fingerprint: wire.key_fingerprint,
8112 signature: wire.signature,
8113 })
8114}
8115
8116fn from_proto_membership(
8122 wire: gunnar_proto::SignedMembershipEvent,
8123) -> ControlResult<gunnar_core::SignedMembershipEvent> {
8124 let event = wire
8125 .event
8126 .ok_or_else(|| ControlError::InvalidArgument("`event` is required".to_owned()))?;
8127 let action = match gunnar_proto::MembershipAction::try_from(event.action) {
8128 Ok(gunnar_proto::MembershipAction::Add) => gunnar_core::MembershipAction::Add,
8129 Ok(gunnar_proto::MembershipAction::Remove) => gunnar_core::MembershipAction::Remove,
8130 _ => {
8131 return Err(ControlError::InvalidArgument(
8132 "`event.action` must be ADD or REMOVE".to_owned(),
8133 ))
8134 }
8135 };
8136 Ok(gunnar_core::SignedMembershipEvent {
8137 event: gunnar_core::MembershipEvent {
8138 group: event.group,
8139 member: event.member,
8140 action,
8141 actor: event.actor,
8142 at_unix_ms: event.at_unix_ms,
8143 },
8144 key_fingerprint: wire.key_fingerprint,
8145 signature: wire.signature,
8146 })
8147}
8148
8149fn proto_action(action: gunnar_core::MembershipAction) -> gunnar_proto::MembershipAction {
8151 match action {
8152 gunnar_core::MembershipAction::Add => gunnar_proto::MembershipAction::Add,
8153 gunnar_core::MembershipAction::Remove => gunnar_proto::MembershipAction::Remove,
8154 }
8155}
8156
8157fn to_proto_event(signed: gunnar_core::SignedEvent) -> gunnar_proto::SignedForgeEvent {
8159 use gunnar_proto::forge_event::Kind;
8160
8161 let kind = match signed.event.event {
8162 gunnar_core::Event::Commented {
8163 body,
8164 path,
8165 line,
8166 blob_oid,
8167 } => Kind::Commented(gunnar_proto::Commented {
8168 body,
8169 path,
8170 line,
8171 blob_oid,
8172 }),
8173 gunnar_core::Event::Approved { commit } => {
8174 Kind::Approved(gunnar_proto::Approved { commit })
8175 }
8176 gunnar_core::Event::Rejected { commit, reason } => {
8177 Kind::Rejected(gunnar_proto::Rejected { commit, reason })
8178 }
8179 gunnar_core::Event::IssueOpened { title, body } => {
8180 Kind::IssueOpened(gunnar_proto::IssueOpened { title, body })
8181 }
8182 gunnar_core::Event::IssueCommented { body } => {
8183 Kind::IssueCommented(gunnar_proto::IssueCommented { body })
8184 }
8185 gunnar_core::Event::IssueClosed { reason } => {
8186 Kind::IssueClosed(gunnar_proto::IssueClosed { reason })
8187 }
8188 };
8189 gunnar_proto::SignedForgeEvent {
8190 event: Some(gunnar_proto::ForgeEvent {
8191 repo: signed.event.repo,
8192 pull: signed.event.pull,
8193 actor: signed.event.actor,
8194 at_unix_ms: signed.event.at_unix_ms,
8195 kind: Some(kind),
8196 }),
8197 key_fingerprint: signed.key_fingerprint,
8198 signature: signed.signature,
8199 }
8200}
8201
8202fn issue_state(repo: &str, issue: &gunnar_core::forge::Issue) -> gunnar_proto::IssueState {
8208 gunnar_proto::IssueState {
8209 repo: repo.to_owned(),
8210 number: issue.number,
8211 title: issue.title.clone(),
8212 body: issue.body.clone(),
8213 author: issue.author.clone(),
8214 open: issue.open,
8215 comments: issue.comments as u64,
8216 opened_at_unix_ms: issue.opened_at_unix_ms,
8217 }
8218}
8219
8220fn from_proto_event(
8225 wire: gunnar_proto::SignedForgeEvent,
8226) -> ControlResult<gunnar_core::SignedEvent> {
8227 use gunnar_proto::forge_event::Kind;
8228
8229 let event = wire
8230 .event
8231 .ok_or_else(|| ControlError::InvalidArgument("`event` is required".to_owned()))?;
8232 let kind = event
8233 .kind
8234 .ok_or_else(|| ControlError::InvalidArgument("`event.kind` is required".to_owned()))?;
8235 let inner = match kind {
8236 Kind::Commented(it) => gunnar_core::Event::Commented {
8237 body: it.body,
8238 path: it.path,
8239 line: it.line,
8240 blob_oid: it.blob_oid,
8241 },
8242 Kind::Approved(it) => gunnar_core::Event::Approved { commit: it.commit },
8243 Kind::Rejected(it) => gunnar_core::Event::Rejected {
8244 commit: it.commit,
8245 reason: it.reason,
8246 },
8247 Kind::IssueOpened(it) => gunnar_core::Event::IssueOpened {
8248 title: it.title,
8249 body: it.body,
8250 },
8251 Kind::IssueCommented(it) => gunnar_core::Event::IssueCommented { body: it.body },
8252 Kind::IssueClosed(it) => gunnar_core::Event::IssueClosed { reason: it.reason },
8253 };
8254 Ok(gunnar_core::SignedEvent {
8255 event: gunnar_core::ForgeEvent {
8256 repo: event.repo,
8257 pull: event.pull,
8258 actor: event.actor,
8259 at_unix_ms: event.at_unix_ms,
8260 event: inner,
8261 },
8262 key_fingerprint: wire.key_fingerprint,
8263 signature: wire.signature,
8264 })
8265}
8266#[cfg(test)]
8267mod tests {
8268 use super::*;
8269
8270 #[test]
8276 fn a_write_on_a_restoring_box_is_unavailable_not_internal() {
8277 let td = tempfile::tempdir().unwrap();
8278 let stores = crate::stores::ServerStores::open(td.path()).unwrap();
8279 stores.txlog().set_follower(true);
8280 let p = gunnar_core::Principal::new("alice", "alice", gunnar_core::PrincipalKind::User).unwrap();
8281 let err = stores.auth().add_principal(p).map_err(internal).expect_err("a follower log takes no write");
8282 assert!(matches!(err, ControlError::Unavailable(_)), "{err:?}");
8283 assert!(err.to_string().contains("restor"), "{err}");
8284 }
8285
8286 #[test]
8296 fn internal_is_refused_by_name_and_the_two_real_visibilities_pass() {
8297 let err = visibility_in(gunnar_proto::Visibility::Internal)
8298 .expect_err("INTERNAL has no implementation and must be refused");
8299 let text = err.to_string();
8300 assert!(
8301 text.contains("internal") && text.contains("private"),
8302 "the refusal must name the value and the real choices: {text}"
8303 );
8304 assert_eq!(
8305 visibility_in(gunnar_proto::Visibility::Public).expect("public is real"),
8306 Visibility::Public
8307 );
8308 assert_eq!(
8309 visibility_in(gunnar_proto::Visibility::Private).expect("private is real"),
8310 Visibility::Private
8311 );
8312 assert_eq!(
8314 visibility_in(gunnar_proto::Visibility::Unspecified).expect("unspecified defaults"),
8315 Visibility::Private
8316 );
8317 }
8318
8319 #[cfg(feature = "replication")]
8331 #[test]
8332 fn abdicate_flush_ceiling_honours_a_sane_override_and_rejects_the_rest() {
8333 assert_eq!(resolve_flush_secs(Some("120")), 120, "a positive override must win");
8335 assert_eq!(resolve_flush_secs(Some(" 90 ")), 90, "surrounding whitespace is trimmed");
8336 assert_eq!(resolve_flush_secs(None), ABDICATE_FLUSH_SECS);
8339 assert_ne!(ABDICATE_FLUSH_SECS, 120, "default must differ from the override arm");
8340 assert_eq!(resolve_flush_secs(Some("")), ABDICATE_FLUSH_SECS, "blank falls back");
8342 assert_eq!(resolve_flush_secs(Some("nope")), ABDICATE_FLUSH_SECS, "non-numeric falls back");
8343 assert_eq!(resolve_flush_secs(Some("0")), ABDICATE_FLUSH_SECS, "zero would truncate — refused");
8344 assert_eq!(resolve_flush_secs(Some("-5")), ABDICATE_FLUSH_SECS, "negative does not parse as u64");
8345 }
8346}
8347
8348fn key_refusal(r: gunnar_core::encryption_key::KeyRefusal) -> ControlError {
8350 use gunnar_core::encryption_key::KeyRefusal as R;
8351 match r {
8352 R::AlreadyHasKey(s) | R::MethodHeld(s) => ControlError::AlreadyExists(s),
8353 R::NoSuchMethod(s) => ControlError::NotFound(s),
8354 R::NoKey(s) => ControlError::FailedPrecondition(s),
8355 R::LastMethod(s) => ControlError::FailedPrecondition(s),
8356 R::Invalid(s) => ControlError::InvalidArgument(s),
8357 R::Store(e) => internal(e),
8358 }
8359}
8360
8361
8362fn method_in(m: gunnar_proto::UnlockMethod) -> ControlResult<gunnar_core::encryption_key::UnlockMethod> {
8363 Ok(gunnar_core::encryption_key::UnlockMethod {
8364 id: String::new(),
8365 kind: gunnar_core::encryption_key::MethodKind::parse(&m.kind).map_err(key_refusal)?,
8366 label: m.label.trim().to_owned(),
8367 method_ref: m.method_ref.trim().to_owned(),
8368 rp_id: m.rp_id.trim().to_owned(),
8369 wrap: m.wrap,
8370 added_ms: 0,
8371 })
8372}
8373
8374fn encryption_key_out(key: Option<gunnar_core::encryption_key::EncryptionKey>) -> gunnar_proto::GetEncryptionKeyResponse {
8375 match key {
8376 None => gunnar_proto::GetEncryptionKeyResponse::default(),
8377 Some(k) => gunnar_proto::GetEncryptionKeyResponse {
8378 exists: true,
8379 public_key: k.public_key,
8380 methods: k
8381 .methods
8382 .into_iter()
8383 .map(|m| gunnar_proto::UnlockMethod {
8384 id: m.id,
8385 kind: m.kind.word().to_owned(),
8386 label: m.label,
8387 method_ref: m.method_ref,
8388 rp_id: m.rp_id,
8389 wrap: m.wrap,
8390 added_ms: m.added_ms,
8391 })
8392 .collect(),
8393 },
8394 }
8395}