1use std::collections::BTreeMap;
103use std::sync::atomic::{AtomicU64, Ordering};
104use std::sync::Arc;
105
106use alkgit::{
107 validate_ref_name, GitRefs, RefLine, RefOutcome, RefUpdate, RepoRecord, StorageError,
108 ATOMIC_PUSH_FAILURE_REASON, FUNNY_REFNAME_REASON,
109};
110use async_trait::async_trait;
111use gix_hash::ObjectId;
112
113use crate::liveness::{fingerprint_rows, RefRow, RefRowState, RefRows, RefRowsSnapshot};
114
115pub const CAS_STALE_REASON: &str = "stale info";
120
121#[derive(Clone, Debug, PartialEq, Eq)]
123enum RowState {
124 Oid(ObjectId),
126 Symref(String),
129}
130
131impl RowState {
132 fn oid(&self) -> Option<ObjectId> {
136 match self {
137 RowState::Oid(oid) => Some(*oid),
138 RowState::Symref(_) => None,
139 }
140 }
141}
142
143enum CasWrite {
146 Set { oid: ObjectId },
147 Delete,
148}
149
150#[derive(Clone, Copy, Debug, PartialEq, Eq)]
153enum CasDenial {
154 FunnyRefname,
156 Stale,
158}
159
160impl CasDenial {
161 fn reason(self) -> &'static str {
162 match self {
163 CasDenial::FunnyRefname => FUNNY_REFNAME_REASON,
164 CasDenial::Stale => CAS_STALE_REASON,
165 }
166 }
167}
168
169fn cas_decide(
182 current: Option<RowState>,
183 update: &RefUpdate,
184) -> Result<Option<CasWrite>, CasDenial> {
185 if validate_ref_name(&update.name).is_err() {
186 return Err(CasDenial::FunnyRefname);
187 }
188 let current_oid = current.as_ref().and_then(RowState::oid);
189 let matched = match (¤t, update.expected) {
190 (None, None) => true,
191 (None, Some(_)) => false,
192 (Some(_), None) => false,
193 (Some(_), Some(_)) => current_oid.is_some() && update.expected == current_oid,
194 };
195 if !matched {
196 return Err(CasDenial::Stale);
197 }
198 Ok(match update.new {
199 Some(new) => Some(CasWrite::Set { oid: new }),
200 None => Some(CasWrite::Delete),
201 })
202}
203
204#[cfg(any(feature = "sqlite", feature = "postgres"))]
210fn row_state_from_columns(
211 repo_id: &str,
212 name: &str,
213 oid_hex: Option<String>,
214 symref_target: Option<String>,
215) -> Result<RowState, StorageError> {
216 match (oid_hex, symref_target) {
217 (Some(hex), None) => {
218 let oid = ObjectId::from_hex(hex.as_bytes()).map_err(|_| {
219 StorageError::Invalid(format!(
220 "git_refs row {repo_id}/{name}: oid_hex is not a hex sha1"
221 ))
222 })?;
223 Ok(RowState::Oid(oid))
224 }
225 (None, Some(target)) => Ok(RowState::Symref(target)),
226 (None, None) => Err(StorageError::Invalid(format!(
227 "git_refs row {repo_id}/{name}: neither oid nor symref target — structurally invalid row state"
228 ))),
229 (Some(_), Some(_)) => Err(StorageError::Invalid(format!(
230 "git_refs row {repo_id}/{name}: both oid and symref target — structurally invalid row state"
231 ))),
232 }
233}
234
235fn row_to_line(name: &str, state: &RowState) -> RefLine {
238 match state {
239 RowState::Oid(oid) => RefLine {
240 name: name.to_owned(),
241 oid: Some(*oid),
242 peeled: None,
243 symref_target: None,
244 },
245 RowState::Symref(target) => RefLine {
246 name: name.to_owned(),
247 oid: None,
248 peeled: None,
249 symref_target: Some(target.clone()),
250 },
251 }
252}
253
254trait CasTable {
263 async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError>;
266
267 async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError>;
269
270 async fn finish(&mut self, abort: bool) -> Result<(), StorageError>;
275}
276
277async fn run_cas<T: CasTable + ?Sized>(
286 table: &mut T,
287 updates: &[RefUpdate],
288 atomic: bool,
289) -> Result<Vec<RefOutcome>, StorageError> {
290 let mut any_denial = false;
291 let mut outcomes = Vec::with_capacity(updates.len());
292 for update in updates {
293 let outcome = match cas_decide(table.state(&update.name).await?, update) {
294 Err(denial) => {
295 any_denial = true;
296 RefOutcome {
297 name: update.name.clone(),
298 result: Err(denial.reason().to_owned()),
299 }
300 }
301 Ok(None) => RefOutcome {
302 name: update.name.clone(),
303 result: Ok(()),
304 },
305 Ok(Some(op)) => {
306 table.write(&update.name, &op).await?;
307 RefOutcome {
308 name: update.name.clone(),
309 result: Ok(()),
310 }
311 }
312 };
313 outcomes.push(outcome);
314 }
315 table.finish(atomic && any_denial).await?;
316 if atomic && any_denial {
317 for outcome in &mut outcomes {
318 if outcome.result.is_ok() {
319 outcome.result = Err(ATOMIC_PUSH_FAILURE_REASON.to_owned());
320 }
321 }
322 }
323 Ok(outcomes)
324}
325
326pub struct InMemoryRefTable {
339 rows: tokio::sync::RwLock<BTreeMap<String, BTreeMap<String, RowState>>>,
340}
341
342impl Default for InMemoryRefTable {
343 fn default() -> Self {
344 Self::new()
345 }
346}
347
348impl InMemoryRefTable {
349 pub fn new() -> Self {
351 InMemoryRefTable {
352 rows: tokio::sync::RwLock::new(BTreeMap::new()),
353 }
354 }
355
356 pub(crate) async fn seed_head(&self, repo_id: &str, target: &str) {
363 self.rows
364 .write()
365 .await
366 .entry(repo_id.to_owned())
367 .or_default()
368 .insert("HEAD".to_owned(), RowState::Symref(target.to_owned()));
369 }
370
371 pub(crate) async fn drop_repo_rows(&self, repo_id: &str) {
376 self.rows.write().await.remove(repo_id);
377 }
378}
379
380struct InMemoryCasTable {
386 working: BTreeMap<String, RowState>,
387 aborted: bool,
388}
389
390impl CasTable for InMemoryCasTable {
391 async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError> {
392 Ok(self.working.get(name).cloned())
393 }
394
395 async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError> {
396 match op {
397 CasWrite::Set { oid } => {
398 self.working.insert(name.to_owned(), RowState::Oid(*oid));
399 }
400 CasWrite::Delete => {
401 self.working.remove(name);
402 }
403 }
404 Ok(())
405 }
406
407 async fn finish(&mut self, abort: bool) -> Result<(), StorageError> {
408 if abort {
409 self.working.clear();
410 self.aborted = true;
411 }
412 Ok(())
413 }
414}
415
416#[async_trait]
417impl GitRefs for InMemoryRefTable {
418 async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
419 let guard = self.rows.read().await;
420 match guard.get(&repo.repo_id) {
421 None => Ok(Vec::new()),
422 Some(tier) => Ok(tier
423 .iter()
424 .map(|(name, state)| row_to_line(name, state))
425 .collect()),
426 }
427 }
428
429 async fn apply_updates(
430 &self,
431 repo: &RepoRecord,
432 updates: Vec<RefUpdate>,
433 atomic: bool,
434 ) -> Result<Vec<RefOutcome>, StorageError> {
435 let mut rows = self.rows.write().await;
436 let mut table = InMemoryCasTable {
437 working: rows.get(&repo.repo_id).cloned().unwrap_or_default(),
438 aborted: false,
439 };
440 let outcomes = run_cas(&mut table, &updates, atomic).await?;
441 if !table.aborted && (!table.working.is_empty() || rows.contains_key(&repo.repo_id)) {
442 rows.insert(repo.repo_id.clone(), table.working);
443 }
444 Ok(outcomes)
445 }
446}
447
448#[cfg(feature = "sqlite")]
466pub struct SqliteRefTable {
467 handle: crate::migrations::CoTenantSqlite,
468}
469
470#[cfg(feature = "sqlite")]
471impl SqliteRefTable {
472 pub fn new(handle: crate::migrations::CoTenantSqlite) -> Self {
475 SqliteRefTable { handle }
476 }
477}
478
479#[cfg(feature = "sqlite")]
480mod sqlite_twin {
481 use super::*;
482
483 struct SqliteCasTable<'a> {
493 tx: Option<rusqlite::Transaction<'a>>,
494 repo_id: String,
495 }
496
497 impl SqliteCasTable<'_> {
498 fn tx(&self) -> Result<&rusqlite::Transaction<'_>, StorageError> {
499 self.tx.as_ref().ok_or_else(|| {
500 StorageError::Io("refs CAS transaction already finalized".to_owned())
501 })
502 }
503
504 fn tx_take(&mut self) -> Result<rusqlite::Transaction<'_>, StorageError> {
505 self.tx.take().ok_or_else(|| {
506 StorageError::Io("refs CAS transaction already finalized".to_owned())
507 })
508 }
509 }
510
511 impl CasTable for SqliteCasTable<'_> {
512 async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError> {
513 let tx = self.tx()?;
514 let queried = tx.query_row(
515 "SELECT oid_hex, symref_target FROM git_refs WHERE repo_id = ?1 AND name = ?2",
516 rusqlite::params![self.repo_id, name],
517 |row| {
518 Ok((
519 row.get::<_, Option<String>>(0)?,
520 row.get::<_, Option<String>>(1)?,
521 ))
522 },
523 );
524 match queried {
525 Ok((oid_hex, symref_target)) => {
526 let state =
527 row_state_from_columns(&self.repo_id, name, oid_hex, symref_target)?;
528 Ok(Some(state))
529 }
530 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
531 Err(e) => Err(storage_io("refs CAS state read", e)),
532 }
533 }
534
535 async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError> {
536 let tx = self.tx()?;
537 match op {
538 CasWrite::Set { oid } => {
539 tx.execute(
540 "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
541 VALUES (?1, ?2, ?3, NULL) \
542 ON CONFLICT(repo_id, name) DO UPDATE SET oid_hex = ?3, symref_target = NULL",
543 rusqlite::params![self.repo_id, name, oid.to_hex().to_string()],
544 )
545 .map_err(|e| storage_io("refs CAS row set", e))?;
546 }
547 CasWrite::Delete => {
548 tx.execute(
549 "DELETE FROM git_refs WHERE repo_id = ?1 AND name = ?2",
550 rusqlite::params![self.repo_id, name],
551 )
552 .map_err(|e| storage_io("refs CAS row delete", e))?;
553 }
554 }
555 Ok(())
556 }
557
558 async fn finish(&mut self, abort: bool) -> Result<(), StorageError> {
559 let tx = self.tx_take()?;
560 if abort {
561 tx.rollback()
562 .map_err(|e| storage_io("refs CAS rollback", e))
563 } else {
564 tx.commit().map_err(|e| storage_io("refs CAS commit", e))
565 }
566 }
567 }
568
569 fn storage_io(context: &str, error: rusqlite::Error) -> StorageError {
570 StorageError::Io(format!("{context}: {error}"))
571 }
572
573 #[async_trait]
574 impl GitRefs for SqliteRefTable {
575 async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
576 let handle = self.handle.clone();
577 let repo_id = repo.repo_id.clone();
578 let listed = tokio::task::spawn_blocking(move || {
579 let conn = handle.lock();
580 let mut stmt = conn
581 .prepare(
582 "SELECT name, oid_hex, symref_target FROM git_refs \
583 WHERE repo_id = ?1 ORDER BY name",
584 )
585 .map_err(|e| storage_io("refs listing prepare", e))?;
586 let rows = stmt
587 .query_map(rusqlite::params![repo_id], |row| {
588 Ok((
589 row.get::<_, String>(0)?,
590 row.get::<_, Option<String>>(1)?,
591 row.get::<_, Option<String>>(2)?,
592 ))
593 })
594 .map_err(|e| storage_io("refs listing query", e))?;
595 let mut lines = Vec::new();
596 for row in rows {
597 let (name, oid_hex, symref_target) =
598 row.map_err(|e| storage_io("refs listing row", e))?;
599 let state = row_state_from_columns(&repo_id, &name, oid_hex, symref_target)?;
600 lines.push(row_to_line(&name, &state));
601 }
602 Ok(lines)
603 })
604 .await
605 .map_err(|e| StorageError::Io(format!("refs listing blocking task: {e}")))?;
606 listed
607 }
608
609 async fn apply_updates(
610 &self,
611 repo: &RepoRecord,
612 updates: Vec<RefUpdate>,
613 atomic: bool,
614 ) -> Result<Vec<RefOutcome>, StorageError> {
615 let handle = self.handle.clone();
616 let repo_id = repo.repo_id.clone();
617 let runtime = tokio::runtime::Handle::current();
618 let applied = tokio::task::spawn_blocking(move || {
619 let mut conn = handle.lock();
620 let tx = conn
621 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
622 .map_err(|e| storage_io("refs CAS BEGIN IMMEDIATE", e))?;
623 let mut table = SqliteCasTable {
624 tx: Some(tx),
625 repo_id,
626 };
627 runtime.block_on(run_cas(&mut table, &updates, atomic))
628 })
629 .await
630 .map_err(|e| StorageError::Io(format!("refs CAS blocking task: {e}")))?;
631 applied
632 }
633 }
634
635 #[async_trait]
640 impl RefRows for SqliteRefTable {
641 async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
642 let handle = self.handle.clone();
643 let rows = tokio::task::spawn_blocking(move || {
644 let conn = handle.lock();
645 let mut stmt = conn
646 .prepare(
647 "SELECT repo_id, name, oid_hex, symref_target FROM git_refs \
648 ORDER BY repo_id, name",
649 )
650 .map_err(|e| storage_io("refs rows snapshot prepare", e))?;
651 let queried = stmt
652 .query_map([], |row| {
653 Ok((
654 row.get::<_, String>(0)?,
655 row.get::<_, String>(1)?,
656 row.get::<_, Option<String>>(2)?,
657 row.get::<_, Option<String>>(3)?,
658 ))
659 })
660 .map_err(|e| storage_io("refs rows snapshot query", e))?;
661 let mut rows = Vec::new();
662 for row in queried {
663 let (repo_id, name, oid_hex, symref_target) =
664 row.map_err(|e| storage_io("refs rows snapshot row", e))?;
665 let state = row_state_from_columns(&repo_id, &name, oid_hex, symref_target)?;
666 rows.push(RefRow {
667 repo_id,
668 name,
669 state: row_state_to_seam(&state),
670 });
671 }
672 Ok(rows)
673 })
674 .await
675 .map_err(|e| StorageError::Io(format!("refs rows snapshot task: {e}")))?;
676 let rows = rows?;
677 Ok(RefRowsSnapshot {
678 fingerprint: fingerprint_rows(&rows),
679 rows,
680 })
681 }
682 }
683}
684
685#[cfg(feature = "postgres")]
718pub struct PgRefTable {
719 handle: crate::postgres::CoTenantPg,
720}
721
722#[cfg(feature = "postgres")]
723impl PgRefTable {
724 pub fn new(handle: crate::postgres::CoTenantPg) -> Self {
727 PgRefTable { handle }
728 }
729}
730
731#[cfg(feature = "postgres")]
732mod pg_twin {
733 use super::*;
734 use crate::migrations::{git_refs_lock_key, qualify};
735 use tokio_postgres::Transaction;
736
737 fn pg_io(context: &str, error: tokio_postgres::Error) -> StorageError {
738 StorageError::Io(format!("{context}: {error}"))
739 }
740
741 struct PgCasTable<'a> {
750 tx: Option<Transaction<'a>>,
751 repo_id: String,
752 schema: String,
753 }
754
755 impl PgCasTable<'_> {
756 fn tx(&self) -> Result<&Transaction<'_>, StorageError> {
757 self.tx.as_ref().ok_or_else(|| {
758 StorageError::Io("refs CAS transaction already finalized".to_owned())
759 })
760 }
761 }
762
763 impl CasTable for PgCasTable<'_> {
764 async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError> {
765 let select = format!(
766 "SELECT oid_hex, symref_target FROM {} \
767 WHERE repo_id = $1 AND name = $2",
768 qualify(&self.schema, "git_refs")
769 );
770 let tx = self.tx()?;
771 match tx.query_opt(&select, &[&self.repo_id, &name]).await {
772 Ok(Some(row)) => {
773 let oid_hex: Option<String> = row
774 .try_get(0)
775 .map_err(|e| pg_io("refs CAS state read", e))?;
776 let symref_target: Option<String> = row
777 .try_get(1)
778 .map_err(|e| pg_io("refs CAS state read", e))?;
779 let state =
780 row_state_from_columns(&self.repo_id, name, oid_hex, symref_target)?;
781 Ok(Some(state))
782 }
783 Ok(None) => Ok(None),
784 Err(e) => Err(pg_io("refs CAS state read", e)),
785 }
786 }
787
788 async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError> {
789 let tx = self.tx()?;
790 match op {
791 CasWrite::Set { oid } => {
792 let upsert = format!(
793 "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
794 VALUES ($1, $2, $3, NULL) \
795 ON CONFLICT (repo_id, name) DO UPDATE \
796 SET oid_hex = $3, symref_target = NULL",
797 qualify(&self.schema, "git_refs")
798 );
799 let oid_hex = oid.to_hex().to_string();
800 tx.execute(&upsert, &[&self.repo_id, &name, &oid_hex])
801 .await
802 .map_err(|e| pg_io("refs CAS row set", e))?;
803 }
804 CasWrite::Delete => {
805 let delete = format!(
806 "DELETE FROM {} WHERE repo_id = $1 AND name = $2",
807 qualify(&self.schema, "git_refs")
808 );
809 tx.execute(&delete, &[&self.repo_id, &name])
810 .await
811 .map_err(|e| pg_io("refs CAS row delete", e))?;
812 }
813 }
814 Ok(())
815 }
816
817 async fn finish(&mut self, abort: bool) -> Result<(), StorageError> {
818 let tx = self.tx.take().ok_or_else(|| {
819 StorageError::Io("refs CAS transaction already finalized".to_owned())
820 })?;
821 if abort {
822 tx.rollback()
823 .await
824 .map_err(|e| pg_io("refs CAS rollback", e))
825 } else {
826 tx.commit().await.map_err(|e| pg_io("refs CAS commit", e))
827 }
828 }
829 }
830
831 #[async_trait]
832 impl GitRefs for PgRefTable {
833 async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
834 let select = format!(
835 "SELECT name, oid_hex, symref_target FROM {} \
836 WHERE repo_id = $1 ORDER BY name",
837 qualify(self.handle.schema(), "git_refs")
838 );
839 let guard = self.handle.lock().await;
840 let rows = guard
841 .query(&select, &[&repo.repo_id])
842 .await
843 .map_err(|e| pg_io("refs listing query", e))?;
844 let mut lines = Vec::with_capacity(rows.len());
845 for row in &rows {
846 let name: String = row.try_get(0).map_err(|e| pg_io("refs listing row", e))?;
847 let oid_hex: Option<String> =
848 row.try_get(1).map_err(|e| pg_io("refs listing row", e))?;
849 let symref_target: Option<String> =
850 row.try_get(2).map_err(|e| pg_io("refs listing row", e))?;
851 let state = row_state_from_columns(&repo.repo_id, &name, oid_hex, symref_target)?;
852 lines.push(row_to_line(&name, &state));
853 }
854 Ok(lines)
855 }
856
857 async fn apply_updates(
858 &self,
859 repo: &RepoRecord,
860 updates: Vec<RefUpdate>,
861 atomic: bool,
862 ) -> Result<Vec<RefOutcome>, StorageError> {
863 let mut guard = self.handle.lock().await;
864 let repo_id = repo.repo_id.clone();
865 let schema = self.handle.schema().to_owned();
866 let tx = guard
867 .transaction()
868 .await
869 .map_err(|e| pg_io("refs CAS BEGIN", e))?;
870 tx.execute(
871 "SELECT pg_advisory_xact_lock($1)",
872 &[&git_refs_lock_key(&schema)],
873 )
874 .await
875 .map_err(|e| pg_io("refs CAS advisory lock", e))?;
876 let mut table = PgCasTable {
877 tx: Some(tx),
878 repo_id,
879 schema,
880 };
881 run_cas(&mut table, &updates, atomic).await
882 }
883 }
884
885 #[async_trait]
890 impl RefRows for PgRefTable {
891 async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
892 let select = format!(
893 "SELECT repo_id, name, oid_hex, symref_target FROM {} \
894 ORDER BY repo_id, name",
895 qualify(self.handle.schema(), "git_refs")
896 );
897 let guard = self.handle.lock().await;
898 let rows = guard
899 .query(&select, &[])
900 .await
901 .map_err(|e| pg_io("refs rows snapshot query", e))?;
902 let mut snapshot = Vec::new();
903 for row in &rows {
904 let repo_id: String = row
905 .try_get(0)
906 .map_err(|e| pg_io("refs rows snapshot row", e))?;
907 let name: String = row
908 .try_get(1)
909 .map_err(|e| pg_io("refs rows snapshot row", e))?;
910 let oid_hex: Option<String> = row
911 .try_get(2)
912 .map_err(|e| pg_io("refs rows snapshot row", e))?;
913 let symref_target: Option<String> = row
914 .try_get(3)
915 .map_err(|e| pg_io("refs rows snapshot row", e))?;
916 let state = row_state_from_columns(&repo_id, &name, oid_hex, symref_target)?;
917 snapshot.push(RefRow {
918 repo_id,
919 name,
920 state: row_state_to_seam(&state),
921 });
922 }
923 drop(guard);
924 Ok(RefRowsSnapshot {
925 fingerprint: fingerprint_rows(&snapshot),
926 rows: snapshot,
927 })
928 }
929 }
930}
931
932fn row_state_to_seam(state: &RowState) -> RefRowState {
942 match state {
943 RowState::Oid(oid) => RefRowState::Oid(*oid),
944 RowState::Symref(target) => RefRowState::Symref(target.clone()),
945 }
946}
947
948#[async_trait]
949impl RefRows for InMemoryRefTable {
950 async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
951 let guard = self.rows.read().await;
952 let rows: Vec<RefRow> = guard
953 .iter()
954 .flat_map(|(repo_id, tier)| {
955 tier.iter().map(move |(name, state)| RefRow {
956 repo_id: repo_id.clone(),
957 name: name.clone(),
958 state: row_state_to_seam(state),
959 })
960 })
961 .collect();
962 Ok(RefRowsSnapshot {
963 fingerprint: fingerprint_rows(&rows),
964 rows,
965 })
966 }
967}
968
969pub struct VersionedRefTable<T: ?Sized> {
978 inner: Arc<T>,
979 version: Arc<AtomicU64>,
980}
981
982impl<T: ?Sized> VersionedRefTable<T> {
983 pub fn new(inner: Arc<T>, version: Arc<AtomicU64>) -> Self {
985 VersionedRefTable { inner, version }
986 }
987
988 pub fn inner(&self) -> &Arc<T> {
991 &self.inner
992 }
993
994 pub fn version(&self) -> u64 {
998 self.version.load(Ordering::Relaxed)
999 }
1000}
1001
1002#[async_trait]
1003impl<T> GitRefs for VersionedRefTable<T>
1004where
1005 T: GitRefs + ?Sized,
1006{
1007 async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
1008 self.inner.list_refs(repo).await
1009 }
1010
1011 async fn apply_updates(
1012 &self,
1013 repo: &RepoRecord,
1014 updates: Vec<RefUpdate>,
1015 atomic: bool,
1016 ) -> Result<Vec<RefOutcome>, StorageError> {
1017 let outcomes = self.inner.apply_updates(repo, updates, atomic).await?;
1028 self.version.fetch_add(1, Ordering::Relaxed);
1029 Ok(outcomes)
1030 }
1031}
1032
1033#[async_trait]
1034impl<T> RefRows for VersionedRefTable<T>
1035where
1036 T: RefRows + ?Sized,
1037{
1038 async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
1039 self.inner.snapshot().await
1040 }
1041}
1042
1043#[cfg(test)]
1044pub(crate) mod battery {
1045 use alkgit::{
1046 GitRefs, RefLine, RefOutcome, RefUpdate, RepoRecord, ATOMIC_PUSH_FAILURE_REASON,
1047 FUNNY_REFNAME_REASON,
1048 };
1049 use gix_hash::ObjectId;
1050
1051 use super::CAS_STALE_REASON;
1052
1053 fn oid(b: u8) -> ObjectId {
1054 ObjectId::from([b; 20])
1055 }
1056
1057 fn up(name: &str, expected: Option<u8>, new: Option<u8>) -> RefUpdate {
1058 RefUpdate {
1059 name: name.to_owned(),
1060 expected: expected.map(oid),
1061 new: new.map(oid),
1062 }
1063 }
1064
1065 fn ok(name: &str) -> RefOutcome {
1066 RefOutcome {
1067 name: name.to_owned(),
1068 result: Ok(()),
1069 }
1070 }
1071
1072 fn deny(name: &str, reason: &str) -> RefOutcome {
1073 RefOutcome {
1074 name: name.to_owned(),
1075 result: Err(reason.to_owned()),
1076 }
1077 }
1078
1079 async fn line_of(table: &dyn GitRefs, repo: &RepoRecord, name: &str) -> Option<RefLine> {
1080 table
1081 .list_refs(repo)
1082 .await
1083 .unwrap()
1084 .into_iter()
1085 .find(|line| line.name == name)
1086 }
1087
1088 pub(super) async fn empty_shapes(table: &dyn GitRefs, repo: &RepoRecord) {
1092 assert!(table.list_refs(repo).await.unwrap().is_empty());
1093 for atomic in [false, true] {
1094 let outcomes = table.apply_updates(repo, Vec::new(), atomic).await.unwrap();
1095 assert!(outcomes.is_empty(), "atomic={atomic}");
1096 }
1097 assert!(table.list_refs(repo).await.unwrap().is_empty());
1098 }
1099
1100 pub(super) async fn stale_shapes_deny_as_outcomes(table: &dyn GitRefs, repo: &RepoRecord) {
1105 table
1106 .apply_updates(repo, vec![up("refs/stale/main", None, Some(1))], false)
1107 .await
1108 .unwrap();
1109
1110 for (shape, update) in [
1111 (
1112 "wrong-expected update",
1113 up("refs/stale/main", Some(2), Some(3)),
1114 ),
1115 ("create over existing", up("refs/stale/main", None, Some(3))),
1116 (
1117 "wrong-expected delete",
1118 up("refs/stale/main", Some(9), None),
1119 ),
1120 ] {
1121 let outcomes = table
1122 .apply_updates(repo, vec![update.clone()], false)
1123 .await
1124 .unwrap();
1125 assert_eq!(
1126 outcomes,
1127 vec![deny("refs/stale/main", CAS_STALE_REASON)],
1128 "{shape} denies as the stale outcome"
1129 );
1130 assert_eq!(
1131 line_of(table, repo, "refs/stale/main").await.unwrap().oid,
1132 Some(oid(1)),
1133 "{shape} left the row untouched"
1134 );
1135 }
1136 }
1137
1138 pub(super) async fn delete_applies_and_force_has_no_fast_forward_check(
1143 table: &dyn GitRefs,
1144 repo: &RepoRecord,
1145 ) {
1146 table
1147 .apply_updates(repo, vec![up("refs/force/main", None, Some(1))], false)
1148 .await
1149 .unwrap();
1150
1151 let outcomes = table
1152 .apply_updates(repo, vec![up("refs/force/main", Some(1), None)], false)
1153 .await
1154 .unwrap();
1155 assert_eq!(outcomes, vec![ok("refs/force/main")]);
1156 assert!(line_of(table, repo, "refs/force/main").await.is_none());
1157
1158 table
1159 .apply_updates(repo, vec![up("refs/force/head", None, Some(4))], false)
1160 .await
1161 .unwrap();
1162 let outcomes = table
1163 .apply_updates(repo, vec![up("refs/force/head", Some(4), Some(5))], false)
1164 .await
1165 .unwrap();
1166 assert_eq!(outcomes, vec![ok("refs/force/head")]);
1167 assert_eq!(
1168 line_of(table, repo, "refs/force/head").await.unwrap().oid,
1169 Some(oid(5)),
1170 "the non-fast-forward new oid applied — no FF check at the policy leg"
1171 );
1172 }
1173
1174 pub(super) async fn non_atomic_reports_per_ref_with_in_call_visibility(
1179 table: &dyn GitRefs,
1180 repo: &RepoRecord,
1181 ) {
1182 table
1183 .apply_updates(repo, vec![up("refs/na/main", None, Some(1))], false)
1184 .await
1185 .unwrap();
1186
1187 let updates = vec![
1188 up("refs/na/main", Some(1), Some(2)),
1189 up("refs/na/topic", Some(3), Some(4)),
1190 up("refs/na/parked", None, Some(5)),
1191 ];
1192 let names = updates.iter().map(|u| u.name.clone()).collect::<Vec<_>>();
1193 let outcomes = table.apply_updates(repo, updates, false).await.unwrap();
1194 assert_eq!(
1195 outcomes.iter().map(|o| o.name.clone()).collect::<Vec<_>>(),
1196 names,
1197 "outcomes arrive in request order"
1198 );
1199 assert_eq!(outcomes[0].result, Ok(()));
1200 assert_eq!(outcomes[1].result, Err(CAS_STALE_REASON.to_owned()));
1201 assert_eq!(outcomes[2].result, Ok(()));
1202 assert_eq!(
1203 line_of(table, repo, "refs/na/main").await.unwrap().oid,
1204 Some(oid(2)),
1205 "the ok applied even though a sibling was denied"
1206 );
1207 assert!(
1208 line_of(table, repo, "refs/na/topic").await.is_none(),
1209 "the denied update applied nothing"
1210 );
1211 assert!(line_of(table, repo, "refs/na/parked").await.is_some());
1212
1213 let outcomes = table
1214 .apply_updates(
1215 repo,
1216 vec![
1217 up("refs/na/inline", None, Some(1)),
1218 up("refs/na/inline", Some(1), Some(2)),
1219 ],
1220 false,
1221 )
1222 .await
1223 .unwrap();
1224 assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
1225 assert_eq!(
1226 line_of(table, repo, "refs/na/inline").await.unwrap().oid,
1227 Some(oid(2)),
1228 "the second update saw the first's applied row (in-call visibility)"
1229 );
1230 }
1231
1232 pub(super) async fn atomic_denials_abort_and_others_report_the_atomic_reason(
1237 table: &dyn GitRefs,
1238 repo: &RepoRecord,
1239 ) {
1240 table
1241 .apply_updates(repo, vec![up("refs/atomic/base", None, Some(1))], false)
1242 .await
1243 .unwrap();
1244
1245 let updates = vec![
1246 up("refs/atomic/base", Some(2), Some(3)),
1247 up("refs/atomic/topic", None, Some(4)),
1248 up("refs/atomic/parked", None, Some(5)),
1249 ];
1250 let outcomes = table.apply_updates(repo, updates, true).await.unwrap();
1251 assert_eq!(outcomes.len(), 3);
1252 assert_eq!(outcomes[0].result, Err(CAS_STALE_REASON.to_owned()));
1253 assert_eq!(
1254 outcomes[1].result,
1255 Err(ATOMIC_PUSH_FAILURE_REASON.to_owned())
1256 );
1257 assert_eq!(
1258 outcomes[2].result,
1259 Err(ATOMIC_PUSH_FAILURE_REASON.to_owned())
1260 );
1261 assert!(
1262 line_of(table, repo, "refs/atomic/topic").await.is_none(),
1263 "the atomic abort left the applied leg discarded"
1264 );
1265 assert!(
1266 line_of(table, repo, "refs/atomic/parked").await.is_none(),
1267 "the atomic abort left the table untouched"
1268 );
1269 assert_eq!(
1270 line_of(table, repo, "refs/atomic/base").await.unwrap().oid,
1271 Some(oid(1))
1272 );
1273
1274 let outcomes = table
1275 .apply_updates(
1276 repo,
1277 vec![
1278 up("refs/atomic/base", Some(1), None),
1279 up("refs/atomic/topic", None, Some(4)),
1280 ],
1281 true,
1282 )
1283 .await
1284 .unwrap();
1285 assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
1286 assert!(line_of(table, repo, "refs/atomic/base").await.is_none());
1287 assert_eq!(
1288 line_of(table, repo, "refs/atomic/topic").await.unwrap().oid,
1289 Some(oid(4)),
1290 "an all-ok atomic call committed whole"
1291 );
1292 table
1293 .apply_updates(repo, vec![up("refs/atomic/topic", Some(4), None)], false)
1294 .await
1295 .unwrap();
1296 }
1297
1298 pub(super) async fn invalid_names_deny_as_outcomes(table: &dyn GitRefs, repo: &RepoRecord) {
1304 table
1305 .apply_updates(repo, vec![up("refs/funny/main", None, Some(1))], false)
1306 .await
1307 .unwrap();
1308
1309 let outcomes = table
1310 .apply_updates(
1311 repo,
1312 vec![
1313 up("refs/funny/main.lock", None, Some(2)),
1314 up("refs/alkgit/host", Some(1), None),
1315 up("refs/funny/main", Some(1), Some(2)),
1316 ],
1317 false,
1318 )
1319 .await
1320 .unwrap();
1321 assert_eq!(
1322 outcomes[0],
1323 deny("refs/funny/main.lock", FUNNY_REFNAME_REASON),
1324 "the .lock deny-list rule denies at the table layer"
1325 );
1326 assert_eq!(
1327 outcomes[1],
1328 deny("refs/alkgit/host", FUNNY_REFNAME_REASON),
1329 "the reserved namespace denies at the table layer"
1330 );
1331 assert_eq!(outcomes[2].result, Ok(()));
1332 assert_eq!(
1333 line_of(table, repo, "refs/funny/main").await.unwrap().oid,
1334 Some(oid(2)),
1335 "the valid ref of the non-atomic call applied"
1336 );
1337
1338 let before = table.list_refs(repo).await.unwrap();
1339 let outcomes = table
1340 .apply_updates(
1341 repo,
1342 vec![
1343 up("refs/funny/main.lock", None, Some(9)),
1344 up("refs/funny/valid", None, Some(3)),
1345 ],
1346 true,
1347 )
1348 .await
1349 .unwrap();
1350 assert_eq!(
1351 outcomes[0],
1352 deny("refs/funny/main.lock", FUNNY_REFNAME_REASON)
1353 );
1354 assert_eq!(
1355 outcomes[1],
1356 deny("refs/funny/valid", ATOMIC_PUSH_FAILURE_REASON)
1357 );
1358 assert_eq!(
1359 table.list_refs(repo).await.unwrap(),
1360 before,
1361 "nothing applied"
1362 );
1363 }
1364
1365 pub(super) async fn unborn_head_lists_the_unborn_line_sorted(
1371 table: &dyn GitRefs,
1372 repo: &RepoRecord,
1373 ) {
1374 assert_eq!(
1375 table.list_refs(repo).await.unwrap(),
1376 vec![RefLine {
1377 name: "HEAD".to_owned(),
1378 oid: None,
1379 peeled: None,
1380 symref_target: Some("refs/heads/main".to_owned()),
1381 }]
1382 );
1383
1384 table
1385 .apply_updates(repo, vec![up("refs/heads/zebra", None, Some(1))], false)
1386 .await
1387 .unwrap();
1388 table
1389 .apply_updates(repo, vec![up("refs/heads/alpaca", None, Some(2))], false)
1390 .await
1391 .unwrap();
1392
1393 let listed = table.list_refs(repo).await.unwrap();
1394 assert_eq!(
1395 listed
1396 .iter()
1397 .map(|line| line.name.as_str())
1398 .collect::<Vec<_>>(),
1399 vec!["HEAD", "refs/heads/alpaca", "refs/heads/zebra"],
1400 "deterministic name-sorted listing, the unborn HEAD first"
1401 );
1402 assert_eq!(listed[0].oid, None);
1403 assert_eq!(listed[0].symref_target.as_deref(), Some("refs/heads/main"));
1404 assert_eq!(listed[1].oid, Some(oid(2)));
1405 assert_eq!(listed[2].oid, Some(oid(1)));
1406 }
1407
1408 pub(super) async fn symref_row_is_cas_unreachable(table: &dyn GitRefs, repo: &RepoRecord) {
1413 for shape in [
1414 up("HEAD", None, Some(1)),
1415 up("HEAD", Some(1), Some(2)),
1416 up("HEAD", Some(1), None),
1417 ] {
1418 let outcomes = table.apply_updates(repo, vec![shape], false).await.unwrap();
1419 assert_eq!(
1420 outcomes,
1421 vec![deny("HEAD", CAS_STALE_REASON)],
1422 "the HEAD symref row never matches a CAS expectation"
1423 );
1424 }
1425 let head = table
1426 .list_refs(repo)
1427 .await
1428 .unwrap()
1429 .into_iter()
1430 .find(|line| line.name == "HEAD")
1431 .expect("the HEAD symref row survived every denied shape");
1432 assert_eq!(head.oid, None);
1433 assert_eq!(head.symref_target.as_deref(), Some("refs/heads/main"));
1434 }
1435}
1436
1437#[cfg(test)]
1438mod tests {
1439 use super::*;
1440 use std::sync::Arc;
1441
1442 use crate::testutil::private_record as repo;
1443 #[cfg(feature = "sqlite")]
1444 use crate::testutil::unique_db_path;
1445
1446 fn oid(b: u8) -> ObjectId {
1447 ObjectId::from([b; 20])
1448 }
1449
1450 #[test]
1451 fn the_denial_and_atomic_reasons_are_the_pinned_texts() {
1452 assert_eq!(CAS_STALE_REASON, "stale info");
1453 assert_eq!(alkgit::ATOMIC_PUSH_FAILURE_REASON, "atomic push failure");
1454 assert_eq!(alkgit::FUNNY_REFNAME_REASON, "funny refname");
1455 }
1456
1457 #[tokio::test]
1460 async fn inmemory_table_carries_the_cas_battery() {
1461 let table: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
1462 let doubled = repo("owner/doubled");
1463
1464 battery::empty_shapes(table.as_ref(), &doubled).await;
1465 battery::stale_shapes_deny_as_outcomes(table.as_ref(), &doubled).await;
1466 battery::delete_applies_and_force_has_no_fast_forward_check(table.as_ref(), &doubled).await;
1467 battery::non_atomic_reports_per_ref_with_in_call_visibility(table.as_ref(), &doubled).await;
1468 battery::atomic_denials_abort_and_others_report_the_atomic_reason(table.as_ref(), &doubled)
1469 .await;
1470 battery::invalid_names_deny_as_outcomes(table.as_ref(), &doubled).await;
1471 }
1472
1473 #[tokio::test]
1477 async fn inmemory_unborn_head_row_lists_and_holds_cas() {
1478 let table = Arc::new(InMemoryRefTable::new());
1479 let unborn = repo("owner/unborn");
1480
1481 table.rows.write().await.insert(
1482 unborn.repo_id.clone(),
1483 [(
1484 "HEAD".to_owned(),
1485 RowState::Symref("refs/heads/main".to_owned()),
1486 )]
1487 .into_iter()
1488 .collect(),
1489 );
1490
1491 battery::unborn_head_lists_the_unborn_line_sorted(table.as_ref(), &unborn).await;
1492 battery::symref_row_is_cas_unreachable(table.as_ref(), &unborn).await;
1493 }
1494
1495 #[tokio::test]
1499 async fn inmemory_committed_delete_all_empties_the_tier() {
1500 let table = InMemoryRefTable::new();
1501 let doubled = repo("owner/delete-all");
1502 table
1503 .apply_updates(
1504 &doubled,
1505 vec![
1506 RefUpdate {
1507 name: "refs/heads/main".to_owned(),
1508 expected: None,
1509 new: Some(oid(1)),
1510 },
1511 RefUpdate {
1512 name: "refs/other".to_owned(),
1513 expected: None,
1514 new: Some(oid(2)),
1515 },
1516 ],
1517 false,
1518 )
1519 .await
1520 .unwrap();
1521 assert_eq!(table.list_refs(&doubled).await.unwrap().len(), 2);
1522
1523 table
1524 .apply_updates(
1525 &doubled,
1526 vec![
1527 RefUpdate {
1528 name: "refs/heads/main".to_owned(),
1529 expected: Some(oid(1)),
1530 new: None,
1531 },
1532 RefUpdate {
1533 name: "refs/other".to_owned(),
1534 expected: Some(oid(2)),
1535 new: None,
1536 },
1537 ],
1538 false,
1539 )
1540 .await
1541 .unwrap();
1542 assert!(
1543 table.list_refs(&doubled).await.unwrap().is_empty(),
1544 "the delete-only call emptied the tier"
1545 );
1546 }
1547
1548 #[cfg(feature = "sqlite")]
1551 #[tokio::test]
1552 async fn sqlite_table_carries_the_cas_battery() {
1553 use crate::migrations::CoTenantSqlite;
1554
1555 let path = unique_db_path("refs", "battery");
1556 let handle = CoTenantSqlite::open(&path).unwrap();
1557 let table: Arc<dyn GitRefs> = Arc::new(SqliteRefTable::new(handle.clone()));
1558 let doubled = repo("owner/doubled");
1559
1560 battery::empty_shapes(table.as_ref(), &doubled).await;
1561 battery::stale_shapes_deny_as_outcomes(table.as_ref(), &doubled).await;
1562 battery::delete_applies_and_force_has_no_fast_forward_check(table.as_ref(), &doubled).await;
1563 battery::non_atomic_reports_per_ref_with_in_call_visibility(table.as_ref(), &doubled).await;
1564 battery::atomic_denials_abort_and_others_report_the_atomic_reason(table.as_ref(), &doubled)
1565 .await;
1566 battery::invalid_names_deny_as_outcomes(table.as_ref(), &doubled).await;
1567
1568 let _ = std::fs::remove_file(&path);
1569 }
1570
1571 #[cfg(feature = "sqlite")]
1574 #[tokio::test]
1575 async fn sqlite_unborn_head_row_lists_and_holds_cas() {
1576 use crate::migrations::CoTenantSqlite;
1577
1578 let path = unique_db_path("refs", "unborn");
1579 let handle = CoTenantSqlite::open(&path).unwrap();
1580 let unborn = repo("owner/unborn");
1581 handle
1582 .lock()
1583 .execute(
1584 "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
1585 VALUES (?1, 'HEAD', NULL, 'refs/heads/main')",
1586 rusqlite::params![unborn.repo_id],
1587 )
1588 .unwrap();
1589 let table: Arc<dyn GitRefs> = Arc::new(SqliteRefTable::new(handle.clone()));
1590
1591 battery::unborn_head_lists_the_unborn_line_sorted(table.as_ref(), &unborn).await;
1592 battery::symref_row_is_cas_unreachable(table.as_ref(), &unborn).await;
1593
1594 let _ = std::fs::remove_file(&path);
1595 }
1596
1597 #[cfg(feature = "sqlite")]
1603 #[tokio::test]
1604 async fn sqlite_two_clashing_applies_from_two_connections() {
1605 use crate::migrations::CoTenantSqlite;
1606
1607 let path = unique_db_path("refs", "clash");
1608 let first =
1609 Arc::new(SqliteRefTable::new(CoTenantSqlite::open(&path).unwrap())) as Arc<dyn GitRefs>;
1610 let second =
1611 Arc::new(SqliteRefTable::new(CoTenantSqlite::open(&path).unwrap())) as Arc<dyn GitRefs>;
1612 let clashers = repo("owner/clash");
1613
1614 let a = tokio::spawn({
1615 let first = first.clone();
1616 let clashers = clashers.clone();
1617 async move {
1618 first
1619 .apply_updates(
1620 &clashers,
1621 vec![RefUpdate {
1622 name: "refs/heads/main".to_owned(),
1623 expected: None,
1624 new: Some(oid(1)),
1625 }],
1626 false,
1627 )
1628 .await
1629 }
1630 });
1631 let b = tokio::spawn({
1632 let second = second.clone();
1633 let clashers = clashers.clone();
1634 async move {
1635 second
1636 .apply_updates(
1637 &clashers,
1638 vec![RefUpdate {
1639 name: "refs/heads/main".to_owned(),
1640 expected: None,
1641 new: Some(oid(2)),
1642 }],
1643 false,
1644 )
1645 .await
1646 }
1647 });
1648 let (a_out, b_out) = tokio::join!(a, b);
1649 let (a_out, b_out) = (a_out.unwrap().unwrap(), b_out.unwrap().unwrap());
1650 let a_won = a_out[0].result.is_ok();
1651 assert!(
1652 a_won != b_out[0].result.is_ok(),
1653 "two clashing creates on one row: exactly one wins"
1654 );
1655 let loser = if a_won {
1656 b_out[0].clone()
1657 } else {
1658 a_out[0].clone()
1659 };
1660 assert_eq!(loser.result, Err(CAS_STALE_REASON.to_owned()));
1661 let winner_oid = if a_won { Some(oid(1)) } else { Some(oid(2)) };
1662 let winner = if a_won { first.clone() } else { second.clone() };
1663 let winner: Arc<dyn GitRefs> = winner;
1664 assert_eq!(
1665 winner.list_refs(&clashers).await.unwrap()[0].oid,
1666 winner_oid,
1667 "the committed winner's row is the row the table carries"
1668 );
1669
1670 drop(first);
1671 drop(second);
1672 let _ = std::fs::remove_file(&path);
1673 }
1674
1675 #[cfg(feature = "sqlite")]
1681 #[tokio::test]
1682 async fn sqlite_corrupted_row_states_map_storage_invalid() {
1683 use crate::migrations::CoTenantSqlite;
1684
1685 let path = unique_db_path("refs", "corrupt");
1686 let handle = CoTenantSqlite::open(&path).unwrap();
1687 let table = SqliteRefTable::new(handle.clone());
1688 let corrupt = repo("owner/corrupt");
1689
1690 handle
1691 .lock()
1692 .execute(
1693 "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
1694 VALUES (?1, 'refs/heads/bent', NULL, NULL)",
1695 rusqlite::params![corrupt.repo_id],
1696 )
1697 .unwrap();
1698 assert!(matches!(
1699 table.list_refs(&corrupt).await.unwrap_err(),
1700 StorageError::Invalid(_)
1701 ));
1702
1703 handle
1704 .lock()
1705 .execute(
1706 "UPDATE git_refs SET oid_hex = ?1, symref_target = 'refs/heads/main' \
1707 WHERE repo_id = ?2 AND name = 'refs/heads/bent'",
1708 rusqlite::params![oid(1).to_string(), corrupt.repo_id],
1709 )
1710 .unwrap();
1711 assert!(matches!(
1712 table.list_refs(&corrupt).await.unwrap_err(),
1713 StorageError::Invalid(_)
1714 ));
1715
1716 handle
1717 .lock()
1718 .execute(
1719 "UPDATE git_refs SET oid_hex = 'not-hex', symref_target = NULL \
1720 WHERE repo_id = ?1 AND name = 'refs/heads/bent'",
1721 rusqlite::params![corrupt.repo_id],
1722 )
1723 .unwrap();
1724 assert!(matches!(
1725 table.list_refs(&corrupt).await.unwrap_err(),
1726 StorageError::Invalid(_)
1727 ));
1728
1729 drop(table);
1730 drop(handle);
1731 let _ = std::fs::remove_file(&path);
1732 }
1733}
1734
1735#[cfg(all(test, feature = "postgres"))]
1743mod pg_twin_tests {
1744 use crate::postgres::live_harness::{drop_schema, live_pg, unique_schema};
1745 use crate::testutil::private_record as repo;
1746 use alkgit::{GitRefs, StorageError};
1747 use gix_hash::ObjectId;
1748
1749 use super::battery;
1750
1751 fn oid(b: u8) -> ObjectId {
1752 ObjectId::from([b; 20])
1753 }
1754
1755 use super::PgRefTable;
1756
1757 #[tokio::test]
1761 async fn pg_table_carries_the_cas_battery() {
1762 let dsn = live_pg!();
1763 let schema = unique_schema("refs-battery");
1764 let table = PgRefTable::new(
1765 crate::postgres::CoTenantPg::open(&dsn, &schema)
1766 .await
1767 .unwrap(),
1768 );
1769 let doubled = repo("owner/doubled");
1770
1771 let table = std::sync::Arc::new(table);
1772 let table: std::sync::Arc<dyn GitRefs> = table;
1773
1774 battery::empty_shapes(table.as_ref(), &doubled).await;
1775 battery::stale_shapes_deny_as_outcomes(table.as_ref(), &doubled).await;
1776 battery::delete_applies_and_force_has_no_fast_forward_check(table.as_ref(), &doubled).await;
1777 battery::non_atomic_reports_per_ref_with_in_call_visibility(table.as_ref(), &doubled).await;
1778 battery::atomic_denials_abort_and_others_report_the_atomic_reason(table.as_ref(), &doubled)
1779 .await;
1780 battery::invalid_names_deny_as_outcomes(table.as_ref(), &doubled).await;
1781
1782 drop(table);
1783 drop_schema(&dsn, &schema).await;
1784 }
1785
1786 #[tokio::test]
1790 async fn pg_unborn_head_row_lists_and_holds_cas() {
1791 let dsn = live_pg!();
1792 let schema = unique_schema("refs-unborn");
1793 let handle = crate::postgres::CoTenantPg::open(&dsn, &schema)
1794 .await
1795 .unwrap();
1796 let unborn = repo("owner/unborn");
1797 handle
1798 .lock()
1799 .await
1800 .execute(
1801 &format!(
1802 "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
1803 VALUES ($1, 'HEAD', NULL, 'refs/heads/main')",
1804 crate::migrations::qualify(&schema, "git_refs")
1805 ),
1806 &[&unborn.repo_id],
1807 )
1808 .await
1809 .unwrap();
1810 let table = std::sync::Arc::new(PgRefTable::new(handle));
1811 let table: std::sync::Arc<dyn GitRefs> = table;
1812
1813 battery::unborn_head_lists_the_unborn_line_sorted(table.as_ref(), &unborn).await;
1814 battery::symref_row_is_cas_unreachable(table.as_ref(), &unborn).await;
1815
1816 drop(table);
1817 drop_schema(&dsn, &schema).await;
1818 }
1819
1820 #[tokio::test]
1826 async fn pg_two_clashing_applies_from_two_connections() {
1827 use super::CAS_STALE_REASON;
1828
1829 let dsn = live_pg!();
1830 let schema = unique_schema("refs-clash");
1831 let first = std::sync::Arc::new(PgRefTable::new(
1832 crate::postgres::CoTenantPg::open(&dsn, &schema)
1833 .await
1834 .unwrap(),
1835 )) as std::sync::Arc<dyn GitRefs>;
1836 let second = std::sync::Arc::new(PgRefTable::new(
1837 crate::postgres::CoTenantPg::open(&dsn, &schema)
1838 .await
1839 .unwrap(),
1840 )) as std::sync::Arc<dyn GitRefs>;
1841 let clashers = repo("owner/clash");
1842
1843 let a = tokio::spawn({
1844 let first = first.clone();
1845 let clashers = clashers.clone();
1846 async move {
1847 first
1848 .apply_updates(
1849 &clashers,
1850 vec![alkgit::RefUpdate {
1851 name: "refs/heads/main".to_owned(),
1852 expected: None,
1853 new: Some(oid(1)),
1854 }],
1855 false,
1856 )
1857 .await
1858 }
1859 });
1860 let b = tokio::spawn({
1861 let second = second.clone();
1862 let clashers = clashers.clone();
1863 async move {
1864 second
1865 .apply_updates(
1866 &clashers,
1867 vec![alkgit::RefUpdate {
1868 name: "refs/heads/main".to_owned(),
1869 expected: None,
1870 new: Some(oid(2)),
1871 }],
1872 false,
1873 )
1874 .await
1875 }
1876 });
1877 let (a_out, b_out) = tokio::join!(a, b);
1878 let (a_out, b_out) = (a_out.unwrap().unwrap(), b_out.unwrap().unwrap());
1879 let a_won = a_out[0].result.is_ok();
1880 assert!(
1881 a_won != b_out[0].result.is_ok(),
1882 "two clashing creates on one row: exactly one wins"
1883 );
1884 let loser = if a_won {
1885 b_out[0].clone()
1886 } else {
1887 a_out[0].clone()
1888 };
1889 assert_eq!(loser.result, Err(CAS_STALE_REASON.to_owned()));
1890 let winner_oid = if a_won { oid(1) } else { oid(2) };
1891 let winner = if a_won { first } else { second };
1892 assert_eq!(
1893 winner.list_refs(&clashers).await.unwrap()[0].oid,
1894 Some(winner_oid),
1895 "the committed winner's row is the row the table carries"
1896 );
1897
1898 drop(winner);
1899 drop_schema(&dsn, &schema).await;
1900 }
1901
1902 #[tokio::test]
1908 async fn pg_corrupted_row_states_map_storage_invalid() {
1909 let dsn = live_pg!();
1910 let schema = unique_schema("refs-corrupt");
1911 let handle = crate::postgres::CoTenantPg::open(&dsn, &schema)
1912 .await
1913 .unwrap();
1914 let table = PgRefTable::new(handle.clone());
1915 let corrupt = repo("owner/corrupt");
1916 let table_name = crate::migrations::qualify(&schema, "git_refs");
1917
1918 let second = crate::postgres::live_harness::connect_direct(&dsn).await;
1919 second
1920 .execute(
1921 &format!(
1922 "INSERT INTO {table_name} (repo_id, name, oid_hex, symref_target) \
1923 VALUES ($1, 'refs/heads/bent', NULL, NULL)"
1924 ),
1925 &[&corrupt.repo_id],
1926 )
1927 .await
1928 .unwrap();
1929 assert!(
1930 matches!(
1931 table.list_refs(&corrupt).await.unwrap_err(),
1932 StorageError::Invalid(_)
1933 ),
1934 "neither column set is structurally invalid"
1935 );
1936
1937 second
1938 .execute(
1939 &format!(
1940 "UPDATE {table_name} SET oid_hex = $1, symref_target = 'refs/heads/main' \
1941 WHERE repo_id = $2 AND name = 'refs/heads/bent'"
1942 ),
1943 &[&oid(1).to_string(), &corrupt.repo_id],
1944 )
1945 .await
1946 .unwrap();
1947 assert!(
1948 matches!(
1949 table.list_refs(&corrupt).await.unwrap_err(),
1950 StorageError::Invalid(_)
1951 ),
1952 "both columns set is structurally invalid"
1953 );
1954
1955 second
1956 .execute(
1957 &format!(
1958 "UPDATE {table_name} SET oid_hex = 'not-hex', symref_target = NULL \
1959 WHERE repo_id = $1 AND name = 'refs/heads/bent'"
1960 ),
1961 &[&corrupt.repo_id],
1962 )
1963 .await
1964 .unwrap();
1965 assert!(matches!(
1966 table.list_refs(&corrupt).await.unwrap_err(),
1967 StorageError::Invalid(_)
1968 ));
1969
1970 drop(second);
1971 drop_schema(&dsn, &schema).await;
1972 }
1973
1974 #[tokio::test]
1980 async fn pg_snapshot_matches_the_reference_impl_canonically() {
1981 use super::InMemoryRefTable;
1982 use crate::liveness::RefRows;
1983
1984 let dsn = live_pg!();
1985 let schema = unique_schema("refs-parity");
1986 let handle = crate::postgres::CoTenantPg::open(&dsn, &schema)
1987 .await
1988 .unwrap();
1989 let seeded = repo("owner/parity");
1990 let sister = repo("owner/parity-sister");
1991
1992 handle
1993 .lock()
1994 .await
1995 .execute(
1996 &format!(
1997 "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
1998 VALUES ($1, 'HEAD', NULL, 'refs/heads/main'), \
1999 ($1, 'refs/heads/main', $2, NULL)",
2000 crate::migrations::qualify(&schema, "git_refs")
2001 ),
2002 &[&seeded.repo_id, &oid(1).to_string()],
2003 )
2004 .await
2005 .unwrap();
2006 handle
2007 .lock()
2008 .await
2009 .execute(
2010 &format!(
2011 "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
2012 VALUES ($1, 'refs/tags/v1', $2, NULL)",
2013 crate::migrations::qualify(&schema, "git_refs")
2014 ),
2015 &[&sister.repo_id, &oid(9).to_string()],
2016 )
2017 .await
2018 .unwrap();
2019
2020 let mem = InMemoryRefTable::new();
2021 {
2022 use super::RowState;
2023 let mut rows = mem.rows.write().await;
2024 rows.insert(
2025 seeded.repo_id.clone(),
2026 [
2027 (
2028 "HEAD".to_owned(),
2029 RowState::Symref("refs/heads/main".to_owned()),
2030 ),
2031 ("refs/heads/main".to_owned(), RowState::Oid(oid(1))),
2032 ]
2033 .into_iter()
2034 .collect(),
2035 );
2036 rows.insert(
2037 sister.repo_id.clone(),
2038 [("refs/tags/v1".to_owned(), RowState::Oid(oid(9)))]
2039 .into_iter()
2040 .collect(),
2041 );
2042 }
2043
2044 let pg = PgRefTable::new(handle);
2045 assert_eq!(
2046 pg.snapshot().await.unwrap(),
2047 mem.snapshot().await.unwrap(),
2048 "the same rows snapshot identically — one canonical form, \
2049 one fingerprint, whatever the engine"
2050 );
2051 assert_eq!(
2052 pg.list_refs(&seeded).await.unwrap(),
2053 mem.list_refs(&seeded).await.unwrap()
2054 );
2055
2056 drop(pg);
2057 drop(mem);
2058 drop_schema(&dsn, &schema).await;
2059 }
2060}