Skip to main content

alkgit_substrate/
refs.rs

1//! The refs co-tenant row table (substrate.md §refs, ADR-019 §2) — the
2//! object-storage family's third trait ([`GitRefs`]) in three shapes:
3//!
4//! - [`InMemoryRefTable`] — the CAS-semantics reference impl, engine-free
5//!   and unconditional in the crate's base (the completeness shape; no
6//!   `mem` feature exists — ADR-025's placement rule).
7//! - [`SqliteRefTable`] (the `sqlite` feature) — the row-CAS twin over
8//!   the co-tenant `git_refs` rows
9//!   ([`CoTenantSqlite`](crate::CoTenantSqlite)): every CAS check and
10//!   apply inside one `BEGIN IMMEDIATE` transaction — the
11//!   one-transaction home that makes `atomic` correct (ADRs 013 §7 / 019
12//!   §2; POC-O verified stale/atomic/delete/force against real `git
13//!   push` behaviors, poco-findings DQ-2).
14//! - `PgRefTable` (the `postgres` feature) — the pg row-CAS twin
15//!   (`crate::postgres::CoTenantPg`): the same one
16//!   transaction per `apply_updates` call, the pg transaction class
17//!   standing in for sqlite's `BEGIN IMMEDIATE` — the call's
18//!   transaction holds the schema's advisory CAS lock (the pg analog
19//!   of the file write lock), so two clashing applies — cross-connection
20//!   included — serialize and the loser sees the committed winner.
21//!
22//! # The CAS semantics (the reference shapes both impls pin)
23//!
24//! Per update, decided in request order against the row state with
25//! in-call visibility (each decision and write sees the updates already
26//! applied within the call); the whole call commits only under the
27//! non-atomic flag — atomic aborts whole on any denial. The wire-path
28//! input always carries distinct ref names per call (duplicate commands
29//! fail closed at the parse — ADR-013 §3), so in practice every update
30//! addresses its own row and the semantics below evaluate cleanly in
31//! the git-observed classes:
32//!
33//! - **Stale** — [`CAS_STALE_REASON`]: the `expected` CAS failed. An
34//!   absent ref never matches an `expected: Some(..)`; a present row
35//!   never matches `expected: None` (create-over-existing) or a
36//!   mismatched oid; a symref row's oid is unborn — it matches nothing
37//!   (the seeded HEAD can be neither created-over nor CAS-deleted —
38//!   git parity: a push never touches a repo's HEAD row).
39//! - **Create** — `expected: None` applies onto an absent name only.
40//! - **Delete** — `new: None` applies onto an oid row whose oid matches
41//!   `expected`. The degenerate `expected: None` + `new: None` shape is
42//!   a no-op success on an absent name (nothing to delete), stale on a
43//!   present one.
44//! - **Force** — no fast-forward check exists (the v1 CAS-only policy,
45//!   ADR-013 §7): an update whose `expected` matches applies with any
46//!   new value, non-fast-forward included (the client's forced-update
47//!   shape is client-displayed; the table's contract is exact-match
48//!   CAS and nothing more).
49//! - **Atomic** — one transaction whole (ADR-013 §10): any denial
50//!   aborts — nothing applies; the denied refs keep their own reasons
51//!   and every other ref of the call reports
52//!   [`ATOMIC_PUSH_FAILURE_REASON`]. An all-passing call commits whole.
53//! - **Non-atomic** — per-update outcomes, each applying when its own
54//!   CAS passes; denials never block the others.
55//!
56//! Per-ref denials are [`RefOutcome`] values, never `Err` (a stale ref
57//! is a business result of receive-pack — ADR-013 §7–8). `Err` is for
58//! substrate failure only: an engine failure rolls the call's
59//! transaction back whole and rides [`StorageError::Io`]
60//! (message-only — no engine source types on the surface, the family
61//! posture).
62//!
63//! # Name validation at the table layer (the pinned mapping)
64//!
65//! Every update's ref name runs [`alkgit::validate_ref_name`] — the
66//! core's `gix_validate::reference` composition plus the
67//! `RESERVED_REF_NAMESPACES` deny-list and `.lock` rule, reused (no
68//! second deny-list; ADR-013 §9's validation homes at the CAS layer —
69//! substrate.md §refs). The mapping this task pins: a rejected name is
70//! a per-ref denial ([`RefOutcome`] with `FUNNY_REFNAME_REASON`),
71//! never `Err`, whatever the row state; under `atomic` the
72//! invalid-name refs keep `funny refname` while the other refs report
73//! `atomic push failure` (the same upstream shape the wire layer's
74//! abort branch pins, ADR-013 §10). The wire layer already validated;
75//! this is the CAS-home defense. `list_refs` does not re-validate:
76//! rows can only arrive through the validated CAS path or the seeding
77//! rule (substrate.md §refs), so read-back names are trusted — but a
78//! structurally invalid row state is [`StorageError::Invalid`]
79//! (mis-state, not a business denial).
80//!
81//! # The row shapes
82//!
83//! `(repo_id, name) → oid, symref_target` rows (POC-O-validated table,
84//! DQ-2). An oid row lists as the plain [`RefLine`]; a symref row (oid
85//! `NULL`, target set — the seeded `HEAD → refs/heads/main`) lists as
86//! the **unborn symref line** `RefLine { oid: None, symref_target:
87//! Some(..) }` — the unborn-HEAD shape `ls-refs=unborn` serves (the
88//! OQ-19 rider (a) fixture falls out of the seeding rule; the seeding
89//! itself lands in `substrate/w6-registry-sqlite`'s `put_repo`, which
90//! writes the HEAD symref row in the record-write transaction — this
91//! task's `list_refs` surfaces the shape on both twins). Listing is
92//! deterministic: sorted by ref name (HEAD sorts before the `refs/`
93//! subtree). `peeled` is never produced by a row table — peel needs
94//! object access the refs surface does not have (ls-refs-arg data).
95//!
96//! Ref-table writes ride their own durable surface — no stream events,
97//! no registry-seq coupling (ADR-019 §7: pushes anchor the event/
98//! watermark legs with the composition-internal mutation marker, the
99//! receive composition's business; ref-table writes are not
100//! registry-family writes).
101
102use 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
115/// The pinned per-ref CAS denial reason (ADR-013 §7–8: CAS-stale
116/// failures carry a server-chosen reason — ours is upstream's
117/// `stale info`-style internal text, the same text the core's contract
118/// doubles pin).
119pub const CAS_STALE_REASON: &str = "stale info";
120
121/// One refs-table row's state as the CAS and `list_refs` read it back.
122#[derive(Clone, Debug, PartialEq, Eq)]
123enum RowState {
124    /// The ordinary ref row: an oid.
125    Oid(ObjectId),
126    /// The symref row (the seeded `HEAD → refs/heads/main`): the unborn
127    /// shape — no oid of its own, a set target.
128    Symref(String),
129}
130
131impl RowState {
132    /// A symref row has no oid of its own — the `None` is exactly the
133    /// unborn property that makes the row never match a CAS
134    /// expectation.
135    fn oid(&self) -> Option<ObjectId> {
136        match self {
137            RowState::Oid(oid) => Some(*oid),
138            RowState::Symref(_) => None,
139        }
140    }
141}
142
143/// The write a passing CAS decision commands (create and update are
144/// the same row write — set the oid; delete removes the row).
145enum CasWrite {
146    Set { oid: ObjectId },
147    Delete,
148}
149
150/// The two denial classes of the CAS check — [`RefOutcome`] reasons,
151/// never errors. The reason text is the respective pinned constant.
152#[derive(Clone, Copy, Debug, PartialEq, Eq)]
153enum CasDenial {
154    /// The name failed the table-layer validation.
155    FunnyRefname,
156    /// The `expected` CAS failed (the [`CAS_STALE_REASON`] class).
157    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
169/// The shared CAS decision — the semantics core both twins run (the
170/// module doc's CAS section is this function's pin). One call per
171/// update, against the update's current row state: the name validation
172/// first (whatever the row state), then the exact-match CAS.
173///
174/// - `expected` must match the row exactly (the CAS contract's
175///   match matrix): absent ⇔ `None`, an oid row ⇔ its own oid only; a
176///   symref row never matches (unborn ⇒ no oid to CAS against).
177/// - A pass commands the write: `new: Some` sets the oid (create and
178///   update are identical at the row level), `new: None` deletes.
179/// - The degenerate `expected: None` + `new: None` shape on an absent
180///   name passes with no write (`None` decision).
181fn 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 (&current, 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/// The row shape's read-back construction from the row's two columns
205/// (the row twins' shared path; the in-memory impl holds the same
206/// shape as an enum). Exactly-one discipline: an oid row is `(oid, no target)`,
207/// a symref row is `(no oid, target)` — anything else is structurally
208/// invalid row state ([`StorageError::Invalid`]).
209#[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
235/// The listed [`RefLine`] of a row (the module doc's row shapes: an oid
236/// row lists plain; the symref row lists the unborn line).
237fn 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
254/// The refs surface's row home behind [`cas_decide`], as both impls
255/// drive it: one call of [`run_cas`], per-update, in request order.
256/// Async surface — the postgres twin is natively async (the
257/// tokio-postgres client has no blocking face); the blocking twins'
258/// method bodies run on their bridged blocking threads (the sqlite
259/// twin polls the whole [`run_cas`] under `Handle::block_on` from one
260/// `spawn_blocking` leg, so its blocking-SQL bodies never touch the
261/// executor).
262trait CasTable {
263    /// The named row's current state (`None` = absent), in-call visible
264    /// (the updates already applied within the call are seen).
265    async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError>;
266
267    /// Apply the commanded CAS write.
268    async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError>;
269
270    /// The call's end transition: commit the whole call (`abort =
271    /// false`) or discard it (`abort` — the atomic denial: nothing the
272    /// call applied survives; the outcomes already carry the denial
273    /// mapping at the caller).
274    async fn finish(&mut self, abort: bool) -> Result<(), StorageError>;
275}
276
277/// The call driver all twins share: one pass over the updates, each
278/// decided and (on pass) applied in request order, then the call's end
279/// transition. Atomic denies map to the pinned reason matrix (denied
280/// refs keep their reasons, the rest report
281/// [`ATOMIC_PUSH_FAILURE_REASON`]); the abort discards every applied
282/// write via [`CasTable::finish`]. An engine failure propagates early —
283/// the impl's transaction drops rolled back and [`StorageError::Io`]
284/// surfaces (substrate failure is not a business outcome).
285async 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
326/// The in-memory refs table — the CAS-semantics reference impl
327/// (engine-free, single-process, stamp-free by construction: it builds
328/// fresh every time and versions nothing durable — ADR-024 §6). Rows
329/// are keyed on the resolved repo id (`repo_id → name → row`), name-sorted
330/// maps (deterministic listing); the whole call runs under one
331/// short-held write lock — the in-memory "transaction" — with the
332/// atomic abort as simply not committing the call's working copy
333/// (per-update decisions inside still see their in-call predecessors).
334/// The lock is the async [`tokio::sync::RwLock`]: its guard is held
335/// across the (never-yielding) CAS futures' awaits and is `Send` —
336/// the [`GitRefs`] `Send`-future surface holds (a blocking guard
337/// across an await would not).
338pub 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    /// A fresh, empty table.
350    pub fn new() -> Self {
351        InMemoryRefTable {
352            rows: tokio::sync::RwLock::new(BTreeMap::new()),
353        }
354    }
355
356    /// The composition's record-transaction seeding leg (substrate.md
357    /// §refs, git parity: `git init --bare` creates the dangling HEAD
358    /// at init, not at first push) — the twins' same-named SQL leg, the
359    /// degenerate shape's row write now that the composition wires this
360    /// table. CAS-unreachable by git parity (a unborn symref row
361    /// matches nothing).
362    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    /// The composition's removal leg (backend.md: repo deletion deletes
372    /// the registry row, the repo's ref rows, and the repo's
373    /// mutation-marker rows together — the twins' same-named SQL leg,
374    /// the degenerate shape's row write).
375    pub(crate) async fn drop_repo_rows(&self, repo_id: &str) {
376        self.rows.write().await.remove(repo_id);
377    }
378}
379
380/// One [`InMemoryRefTable`] call's working copy — the in-memory
381/// "transaction": decisions and writes land on the copy taken at call
382/// start; [`CasTable::finish`] marks the abort. The write-back happens
383/// only on a committed call (never an abort, never a substrate `Err` —
384/// an early error return drops the working copy whole).
385struct 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/// The sqlite row-CAS twin over the co-tenant `git_refs` rows
449/// (substrate.md §refs, ADR-019 §2) — the `atomic` realization as the
450/// one-transaction home: every [`GitRefs::apply_updates`] call runs all
451/// its CAS check-and-applies inside one `BEGIN IMMEDIATE` transaction
452/// on the co-tenant connection (the migration-checked schema from
453/// `substrate/w6-migrations`), so two clashing applies — from this
454/// process or a co-tenant connection of another — serialize on the row
455/// write lock and the loser sees the committed winner (`stale info`),
456/// while an atomic denial rolls its transaction back whole, leaving the
457/// rows exactly as they were (POC-O's demonstrated shapes, DQ-2).
458///
459/// Blocking SQL bridges off the async executor
460/// (`tokio::task::spawn_blocking`) per the composition's concurrency
461/// model; the co-tenant handle's pinned busy timeout absorbs the
462/// cross-process write-lock waits. Construction takes the
463/// already-migrated handle ([`CoTenantSqlite::open`](crate::migrations::CoTenantSqlite::open)
464/// runs the migrations) — no construction I/O here.
465#[cfg(feature = "sqlite")]
466pub struct SqliteRefTable {
467    handle: crate::migrations::CoTenantSqlite,
468}
469
470#[cfg(feature = "sqlite")]
471impl SqliteRefTable {
472    /// Mount the refs row CAS over a co-tenant handle (the same handle
473    /// the sqlite registry twin consumes).
474    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    /// One sqlite [`GitRefs::apply_updates`] call's in-transaction
484    /// plumbing: the `BEGIN IMMEDIATE` transaction taken up front, the
485    /// decisions and row writes inside it, the commit-or-rollback at
486    /// [`CasTable::finish`]. An engine failure at any leg propagates
487    /// out of [`run_cas`]; the dropped transaction rolls the whole call
488    /// back (substrate failure, not a business outcome). The methods'
489    /// bodies are the blocking rusqlite calls — polled on the
490    /// blocking thread under the [`run_cas`] bridge, never the
491    /// executor.
492    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    /// The liveness seam's row read: one ordered query over the
636    /// co-tenant rows (`ORDER BY repo_id, name` — the canonical
637    /// order the fingerprint folds), bridged off the executor like
638    /// every blocking-SQL leg.
639    #[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/// The postgres row-CAS twin over the co-tenant `git_refs` rows
686/// (substrate.md §refs, ADR-019 §3 — the fleet/durable-shared engine
687/// assembly's `GitRefs`, the `postgres` feature; the wave-6 gate's N-c
688/// rider). The [`SqliteRefTable`] semantics translated engine-swapped
689/// over [`CoTenantPg`](crate::postgres::CoTenantPg): every
690/// [`GitRefs::apply_updates`] call runs all its CAS check-and-applies
691/// inside **one transaction**, the pg transaction class standing in
692/// for sqlite's `BEGIN IMMEDIATE`.
693///
694/// The transaction's serialization home: the call takes the schema's
695/// advisory CAS lock (`pg_advisory_xact_lock` — the pg analog of the
696/// sqlite file's write lock) as the transaction's first statement, so
697/// two clashing applies — cross-connection included — serialize
698/// before their reads: the loser proceeds only after the winner's
699/// commit and sees committed state, which under the shared
700/// `cas_decide` matrix is exactly the [`CAS_STALE_REASON`] denial.
701/// (The same-handle shape serializes on the co-tenant client mutex
702/// first anyway; the advisory lock is the whole-fleet home.) Within
703/// the transaction, the decisions ride `READ COMMITTED` reads of the
704/// co-tenant rows with in-call visibility — the CAS loop's own
705/// applies are visible because they are this transaction's writes.
706///
707/// Registry-family writes are untouched (the `git_refs` rows' other
708/// writers — the record transactions' HEAD seed/removal legs — run
709/// their own transactions; ref-table writes are not registry-family
710/// writes, ADR-019 §7: no `registry_seq` coupling here).
711///
712/// Natively async — every leg is a straight await on the co-tenant
713/// handle's client (the tokio-postgres model; the blocking bridge is
714/// the *other* engines'). Construction takes the already-migrated
715/// handle ([`CoTenantPg::open`](crate::postgres::CoTenantPg::open)
716/// runs the migrations) — no construction I/O here.
717#[cfg(feature = "postgres")]
718pub struct PgRefTable {
719    handle: crate::postgres::CoTenantPg,
720}
721
722#[cfg(feature = "postgres")]
723impl PgRefTable {
724    /// Mount the refs row CAS over a co-tenant handle (the same handle
725    /// the pg registry twin consumes; one instance serves both).
726    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    /// One pg [`GitRefs::apply_updates`] call's in-transaction
742    /// plumbing: the transaction taken (and the advisory CAS lock
743    /// held) by the caller up front, the decisions and row writes
744    /// inside it, the commit-or-rollback at [`CasTable::finish`]. An
745    /// engine failure at any leg propagates out of [`run_cas`]; the
746    /// dropped transaction rolls the whole call back (the
747    /// tokio-postgres transaction's drop performs it — substrate
748    /// failure, not a business outcome).
749    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    /// The liveness seam's row read: one ordered query over the
886    /// co-tenant rows (`ORDER BY repo_id, name` — the canonical
887    /// order the fingerprint folds), a straight await (natively
888    /// async).
889    #[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
932// The liveness seam's row reads (crate::liveness — the ref-table
933// closure's durable state): a plain ordered read of the rows the CAS
934// writes, plus the canonical fingerprint over them. Both tables carry
935// it; the write-counter wrapper forwards.
936//
937// The fingerprint's purpose (memo invalidation, cross-process) lives
938// at the seam: this is the durable-marker option realized read-side
939// — no schema change, the rows themselves are the marker.
940
941fn 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
969/// The refs write counter — the liveness source's in-process
970/// invalidation half (crate::liveness "the arbitration window"):
971/// bumped on every CAS call through the composition's refs impl, read
972/// by the source's arbitration consult. See [`VersionedRefTable`].
973///
974/// The wave-6 refs rows carry no version counter; the wrapper is the
975/// counter's home (the POC-O table-embedded counter's shape, moved
976/// into the composition wiring — the tables stay row-homes).
977pub struct VersionedRefTable<T: ?Sized> {
978    inner: Arc<T>,
979    version: Arc<AtomicU64>,
980}
981
982impl<T: ?Sized> VersionedRefTable<T> {
983    /// The wrapper over the table and the shared write counter.
984    pub fn new(inner: Arc<T>, version: Arc<AtomicU64>) -> Self {
985        VersionedRefTable { inner, version }
986    }
987
988    /// The wrapped table (the rows' home — the liveness snapshot
989    /// forwards through it).
990    pub fn inner(&self) -> &Arc<T> {
991        &self.inner
992    }
993
994    /// The write counter's current value (the liveness source's
995    /// arbitration epoch signal — the [`GitRefs`] impl's verbatim
996    /// delegation adds only the bump).
997    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        // The epoch discipline's bump-after-commit half (the
1018        // [`crate::liveness`] docs): the mark consult's counter load
1019        // rides ahead of its snapshot read, so every write's bump must
1020        // land strictly after its committed rows — the only pairing
1021        // that keeps a memo from pairing an included bump with a
1022        // snapshot read that missed the write's rows. Every green call
1023        // bumps — the per-ref denials ride the outcome list, and an
1024        // all-denied call landing nothing still moves the counter past
1025        // any consult racing it (over-invalidation, the safe
1026        // direction); only `Err` bumps nothing.
1027        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    /// The empty-set shapes: an untouched table lists empty; an empty
1089    /// update call is an empty outcome set (both flags) and nothing
1090    /// else.
1091    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    /// The stale classes (ADR-013 §7): a wrong-`expected` update, a
1101    /// create-over-existing, a wrong-`expected` delete — each a denied
1102    /// [`RefOutcome`] (the pinned `stale info` reason), never `Err`,
1103    /// with the row untouched.
1104    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    /// Delete and the force shape: an `expected`-matched delete applies
1139    /// (the row is gone); a non-fast-forward update applies with no
1140    /// fast-forward check (the v1 CAS-only policy, ADR-013 §7) — the
1141    /// unrelated new value lands on a matching `expected`.
1142    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    /// Non-atomic (ADR-018 §6): per-update outcomes in request order —
1175    /// each ok applies, denials never block the others; plus the
1176    /// in-call visibility: two updates to one name in one call see
1177    /// each other.
1178    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    /// Atomic denial (ADR-013 §10): nothing applied — the denied refs
1233    /// keep their own reasons and the other refs of the call report
1234    /// `atomic push failure`; an all-ok atomic call commits whole
1235    /// (delete + create in one transaction).
1236    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    /// The table-layer name validation's pinned mapping (ADR-013 §9/
1299    /// §10, substrate.md §refs): a rejected name denies the ref with
1300    /// `funny refname` — an outcome, never `Err`, whatever the row
1301    /// state; under atomic the funny refs keep their reason while the
1302    /// other refs report `atomic push failure` and nothing applies.
1303    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    /// The seeded `HEAD → refs/heads/main` symref row lists as the
1366    /// unborn line (`oid: None`, the target set) — the row shape
1367    /// `ls-refs=unborn` serves (the OQ-19 rider (a) fixture),
1368    /// name-sorted before the `refs/` rows; the listing stays
1369    /// deterministic as rows are added.
1370    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    /// The symref row is CAS-unreachable: create-over, update, and
1409    /// delete against `HEAD` all deny stale (a symref row's oid is
1410    /// unborn — it matches nothing; git parity: pushes never touch the
1411    /// repo's HEAD row).
1412    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    /// The battery over the engine-free reference impl — the CAS
1458    /// semantics this task pins (ADR-013 §7/§10's shapes).
1459    #[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    /// The unborn-HEAD shapes over the seeded row (each test writes the
1474    /// HEAD symref row directly — the `substrate/w6-registry-sqlite`
1475    /// seeding rule's row shape; put_repo's seeding is not this task's).
1476    #[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    /// Non-atomic deletes clear the call's own tier too — a committed
1496    /// delete-only call's working copy replaces the stored rows (a
1497    /// delete-emptied repo lists empty, never stale leftovers).
1498    #[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    /// The battery over the sqlite row-CAS twin — the same semantics in
1549    /// rows, one `BEGIN IMMEDIATE` transaction per call.
1550    #[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    /// The unborn-HEAD fixture over the seeded rows (the seeding rule's
1572    /// row shape seeded directly).
1573    #[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    /// POC-O's demonstrated contention shape (DQ-2): two clashing
1598    /// applies from two co-tenant connections of one file — the loser
1599    /// sees the committed winner (`stale info`); the transactions
1600    /// serialize on the row write lock, never erroring (pinned 5000 ms
1601    /// busy timeout).
1602    #[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    /// The corrupted-row defenses: neither column set, both columns
1676    /// set, or a non-hex oid — structurally invalid row state maps
1677    /// [`StorageError::Invalid`] on listing. The in-memory impl carries
1678    /// the same shape as a two-arm enum — the mis-state is
1679    /// unexpressible there by construction.
1680    #[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/// The pg twin's battery — the same CAS script the in-memory reference
1736/// and the sqlite twin run, over the live-pg harness (the
1737/// `ALKBLOBS_PG_*`/`ALKGIT_SUBSTRATE_PG_*` convention — one container,
1738/// one convention; clean skip when the host env is absent, so
1739/// `--all-features` stays green everywhere). Every test works in a
1740/// unique, freshly-migrated PostgreSQL schema and drops it after — the
1741/// harness database stays untouched.
1742#[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    /// The battery over the pg row-CAS twin — the semantics pinned on
1758    /// the reference impl, re-run engine-swapped (no drift possible:
1759    /// the same shared `cas_decide` + `run_cas` decide both).
1760    #[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    /// The unborn-HEAD fixture over the seeded row (the seeding rule's
1787    /// row shape, seeded directly — put_repo's seeding is not this
1788    /// task's; the pg twin's migrations already carry the table).
1789    #[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    /// POC-O's demonstrated contention shape (DQ-2) over pg: two
1821    /// clashing applies from two co-tenant connections of one schema —
1822    /// the advisory CAS lock serializes the transactions before their
1823    /// reads, the loser sees the committed winner (`stale info`), and
1824    /// neither errors.
1825    #[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    /// The corrupted-row defenses: neither column set, both columns
1903    /// set, or a non-hex oid — structurally invalid row state maps
1904    /// [`StorageError::Invalid`] on listing (the mis-states are the pg
1905    /// rows' expressible shape; the in-memory two-arm enum cannot
1906    /// express them).
1907    #[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    /// The parity discipline (the wave-6 posture extended to the refs
1975    /// twin): the same canonical row set snapshots identically over
1976    /// the pg twin and the in-memory reference — rows and the
1977    /// fingerprint both (the liveness memo's cross-engine rebuild key
1978    /// is one value whatever the engine).
1979    #[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}