#[cfg(feature = "sqlite")]
use crate::error::OpenError;
#[cfg(feature = "sqlite")]
mod sqlite_twin {
use super::OpenError;
use std::path::Path;
use std::sync::Arc;
const STAMP_TABLE: &str = "alkgit_schema_stamp";
pub(crate) const BUSY_TIMEOUT_MS: i64 = 5_000;
pub(crate) const STAMP_TABLE_BOOTSTRAP: &str = "
CREATE TABLE IF NOT EXISTS alkgit_schema_stamp (
id INTEGER PRIMARY KEY CHECK (id = 1),
stamp INTEGER NOT NULL
);
INSERT OR IGNORE INTO alkgit_schema_stamp (id, stamp) VALUES (1, 0);";
pub(crate) struct Step {
pub(crate) version: i64,
pub(crate) name: &'static str,
pub(crate) sql: &'static str,
pub(crate) verify_sql: &'static str,
}
const SQLITE_STEPS: &[Step] = &[Step {
version: 1,
name: "initial co-tenant schema (records, refs, seq, mutation markers, LFS liveness rows)",
sql: "CREATE TABLE git_repos (
repo_id TEXT PRIMARY KEY,
visibility TEXT NOT NULL CHECK (visibility IN ('public', 'private')),
grants TEXT NOT NULL,
seq INTEGER NOT NULL
);
CREATE TABLE git_refs (
repo_id TEXT NOT NULL,
name TEXT NOT NULL,
oid_hex TEXT,
symref_target TEXT,
PRIMARY KEY (repo_id, name)
);
CREATE TABLE registry_seq (
id INTEGER PRIMARY KEY CHECK (id = 1),
value INTEGER NOT NULL
);
INSERT INTO registry_seq (id, value) VALUES (1, 0);
CREATE TABLE repo_mutation_markers (
repo_id TEXT PRIMARY KEY,
seq INTEGER NOT NULL
);
CREATE TABLE lfs_objects (
repo_id TEXT NOT NULL,
oid TEXT NOT NULL,
size INTEGER NOT NULL,
refcount INTEGER NOT NULL,
PRIMARY KEY (repo_id, oid)
);",
verify_sql: "SELECT count(*) = 5 FROM sqlite_master WHERE type = 'table' \
AND name IN ('git_repos', 'git_refs', 'registry_seq', \
'repo_mutation_markers', 'lfs_objects')",
}];
#[derive(Clone)]
pub struct CoTenantSqlite {
conn: Arc<parking_lot::Mutex<rusqlite::Connection>>,
}
impl CoTenantSqlite {
pub fn open(path: impl AsRef<Path>) -> Result<Self, OpenError> {
let path = path.as_ref();
if path == Path::new(":memory:") {
return Err(OpenError::Invalid(
"sqlite memory mode confers no co-tenancy — compose the \
mem engines for ephemeral assemblies"
.into(),
));
}
let conn = rusqlite::Connection::open(path).map_err(|e| {
OpenError::Engine(format!("co-tenant sqlite open ({}): {e}", path.display()))
})?;
conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS)
.map_err(|e| OpenError::Engine(format!("co-tenant sqlite busy_timeout: {e}")))?;
if let Err(e) = conn.pragma_update(None, "journal_mode", "WAL") {
const VERIFY_RETRIES: usize = 25;
let mut became_wal = false;
for _ in 0..VERIFY_RETRIES {
let mode: Result<String, _> =
conn.query_row("PRAGMA journal_mode", [], |r| r.get(0));
if mode.is_ok_and(|m| m.eq_ignore_ascii_case("wal")) {
became_wal = true;
break;
}
if conn.pragma_update(None, "journal_mode", "WAL").is_ok() {
became_wal = true;
break;
}
std::thread::sleep(std::time::Duration::from_millis(10));
}
if !became_wal {
return Err(OpenError::Engine(format!(
"co-tenant sqlite journal_mode: {e}"
)));
}
}
conn.pragma_update(None, "synchronous", "NORMAL")
.map_err(|e| OpenError::Engine(format!("co-tenant sqlite synchronous: {e}")))?;
let mut conn = conn;
migrate(&mut conn)?;
Ok(CoTenantSqlite {
conn: Arc::new(parking_lot::Mutex::new(conn)),
})
}
pub fn lock(&self) -> parking_lot::MutexGuard<'_, rusqlite::Connection> {
self.conn.lock()
}
pub fn stamp(&self) -> Result<i64, OpenError> {
read_stamp(&self.lock())
}
pub fn step_count() -> usize {
SQLITE_STEPS.len()
}
}
pub(crate) fn migrate(conn: &mut rusqlite::Connection) -> Result<i64, OpenError> {
run_steps(conn, SQLITE_STEPS)
}
pub(crate) fn run_steps(
conn: &mut rusqlite::Connection,
steps: &[Step],
) -> Result<i64, OpenError> {
conn.execute_batch(STAMP_TABLE_BOOTSTRAP)
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("stamp-table bootstrap ({STAMP_TABLE}): {e}"),
})?;
for step in steps {
apply_step(conn, step)?;
}
read_stamp(conn)
}
fn apply_step(conn: &mut rusqlite::Connection, step: &Step) -> Result<(), OpenError> {
let step_no = step.version as usize;
let tx = conn
.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("step {step_no} ({}): BEGIN IMMEDIATE: {e}", step.name),
})?;
let stamp_now: i64 = tx
.query_row(read_stamp_sql().as_str(), [], |r| r.get(0))
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("step {step_no} ({}): stamp read: {e}", step.name),
})?;
if stamp_now >= step.version {
return Ok(());
}
let outcome = tx
.execute_batch(step.sql)
.and_then(|_| tx.execute(write_stamp_sql().as_str(), rusqlite::params![step.version]))
.and_then(|_| tx.commit());
match outcome {
Ok(()) => Ok(()),
Err(e) => verify_race(conn, step, &e),
}
}
fn verify_race(
conn: &rusqlite::Connection,
step: &Step,
failure: &rusqlite::Error,
) -> Result<(), OpenError> {
if step_effect_present(conn, step)? {
let step_no = step.version as usize;
conn.execute(mark_stamp_sql().as_str(), rusqlite::params![step.version])
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!(
"step {step_no} ({}): race-swallow stamp heal: {e}",
step.name
),
})?;
Ok(())
} else {
let step_no = step.version as usize;
Err(OpenError::Migration {
step: step_no,
reason: format!("step {step_no} ({}): {failure}", step.name),
})
}
}
fn step_effect_present(conn: &rusqlite::Connection, step: &Step) -> Result<bool, OpenError> {
let step_no = step.version as usize;
match conn.query_row(step.verify_sql, [], |r| r.get::<_, i64>(0)) {
Ok(truthy) => Ok(truthy > 0),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(false),
Err(e) => Err(OpenError::Migration {
step: step_no,
reason: format!("step {step_no} ({}): verification query: {e}", step.name),
}),
}
}
fn read_stamp_sql() -> String {
format!("SELECT stamp FROM {STAMP_TABLE} WHERE id = 1")
}
fn write_stamp_sql() -> String {
format!("UPDATE {STAMP_TABLE} SET stamp = ?1 WHERE id = 1")
}
fn mark_stamp_sql() -> String {
format!("UPDATE {STAMP_TABLE} SET stamp = ?1 WHERE id = 1 AND stamp < ?1")
}
pub(crate) fn read_stamp(conn: &rusqlite::Connection) -> Result<i64, OpenError> {
conn.query_row(read_stamp_sql().as_str(), [], |r| r.get(0))
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("stamp read ({STAMP_TABLE}): {e}"),
})
}
#[cfg(test)]
pub(crate) fn presence_sql(table: &str) -> &'static str {
Box::leak(
format!(
"SELECT count(*) = 1 FROM sqlite_master \
WHERE type = 'table' AND name = '{table}'"
)
.into_boxed_str(),
)
}
#[cfg(test)]
pub(crate) fn step_probe(table: &str, name: &'static str, version: i64) -> Step {
Step {
version,
name,
sql: Box::leak(format!("CREATE TABLE {table} (x TEXT, y TEXT)").into_boxed_str()),
verify_sql: presence_sql(table),
}
}
#[cfg(test)]
pub(crate) fn failing_step_probe(table: &str, name: &'static str, version: i64) -> Step {
Step {
version,
name,
sql: Box::leak(
format!(
"CREATE TABLE {table} (x TEXT, y TEXT); CREATE TABLE {table} (x TEXT, y TEXT)"
)
.into_boxed_str(),
),
verify_sql: presence_sql(table),
}
}
}
#[cfg(feature = "sqlite")]
pub use sqlite_twin::CoTenantSqlite;
#[cfg(feature = "postgres")]
pub(crate) mod pg_twin {
use std::sync::Arc;
use alkstore_postgres::quote_identifier;
use tokio_postgres::NoTls;
use crate::error::OpenError;
pub const DEFAULT_SCHEMA: &str = "alkgit";
pub(crate) const STAMP_TABLE: &str = "alkgit_schema_stamp";
pub(crate) const STEP_VERSIONS: &[i64] = &[1];
fn validated_schema(schema: &str) -> Result<Arc<str>, OpenError> {
let schema = if schema.is_empty() {
DEFAULT_SCHEMA
} else {
schema
};
if schema.len() > 63
|| schema.starts_with(|c: char| c.is_ascii_digit())
|| schema
.chars()
.any(|c| c.is_control() || c == '"' || c == '\0')
{
return Err(OpenError::Invalid(format!(
"postgres co-tenant schema name is not a usable identifier: {schema:?}"
)));
}
Ok(Arc::from(schema))
}
pub(crate) fn qualify(schema: &str, table: &str) -> String {
format!("{}.{}", quote_identifier(schema), quote_identifier(table))
}
fn advisory_lock_key(scope: &str) -> i64 {
let mut hash: u64 = 0xcbf29ce484222325;
for byte in scope.as_bytes() {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x100000001b3);
}
hash as i64
}
fn migration_lock_key(schema: &str) -> i64 {
advisory_lock_key(&format!("{schema}\0{STAMP_TABLE}"))
}
pub(crate) fn git_refs_lock_key(schema: &str) -> i64 {
advisory_lock_key(&format!("{schema}\0git_refs"))
}
pub(crate) fn lfs_objects_lock_key(schema: &str) -> i64 {
advisory_lock_key(&format!("{schema}\0lfs_objects"))
}
#[derive(Clone)]
pub struct CoTenantPg {
client: Arc<tokio::sync::Mutex<tokio_postgres::Client>>,
schema: Arc<str>,
}
impl CoTenantPg {
pub async fn open(config: &str, schema: &str) -> Result<Self, OpenError> {
let schema = validated_schema(schema)?;
let cfg = config
.parse::<tokio_postgres::Config>()
.map_err(|e| OpenError::Engine(format!("co-tenant postgres config: {e}")))?;
let (client, conn) = cfg
.connect(NoTls)
.await
.map_err(|e| OpenError::Engine(format!("co-tenant postgres connect: {e}")))?;
tokio::spawn(async move {
let _ = conn.await;
});
let mut client = client;
migrate(&mut client, &schema).await?;
Ok(CoTenantPg {
client: Arc::new(tokio::sync::Mutex::new(client)),
schema,
})
}
pub fn schema(&self) -> &str {
&self.schema
}
pub async fn lock(&self) -> tokio::sync::MutexGuard<'_, tokio_postgres::Client> {
self.client.lock().await
}
pub async fn stamp(&self) -> Result<i64, OpenError> {
let read = format!(
"SELECT stamp FROM {} WHERE id = 1",
qualify(&self.schema, STAMP_TABLE)
);
let guard = self.client.lock().await;
stamp_of(&guard, &read).await
}
pub fn step_count() -> usize {
STEP_VERSIONS.len()
}
}
async fn stamp_of<R: tokio_postgres::ToStatement>(
client: &tokio_postgres::Client,
read: &R,
) -> Result<i64, OpenError> {
client
.query_one(read, &[])
.await
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("stamp read ({STAMP_TABLE}): {e}"),
})
.and_then(|row| {
row.try_get::<_, i64>(0).map_err(|e| OpenError::Migration {
step: 0,
reason: format!("stamp read ({STAMP_TABLE}): {e}"),
})
})
}
pub(crate) struct PgStep {
pub(crate) version: i64,
name: &'static str,
sql: String,
tables: &'static [&'static str],
}
pub(crate) fn build_steps(schema: &str) -> Vec<PgStep> {
let q = |table: &str| qualify(schema, table);
vec![PgStep {
version: 1,
name:
"initial co-tenant schema (records, refs, seq, mutation markers, LFS liveness rows)",
sql: format!(
"CREATE TABLE {} (
repo_id TEXT PRIMARY KEY,
visibility TEXT NOT NULL CHECK (visibility IN ('public', 'private')),
grants TEXT NOT NULL,
seq BIGINT NOT NULL
);
CREATE TABLE {} (
repo_id TEXT NOT NULL,
name TEXT NOT NULL,
oid_hex TEXT,
symref_target TEXT,
PRIMARY KEY (repo_id, name)
);
CREATE TABLE {} (
id INTEGER PRIMARY KEY CHECK (id = 1),
value BIGINT NOT NULL
);
INSERT INTO {} (id, value) VALUES (1, 0);
CREATE TABLE {} (
repo_id TEXT PRIMARY KEY,
seq BIGINT NOT NULL
);
CREATE TABLE {} (
repo_id TEXT NOT NULL,
oid TEXT NOT NULL,
size BIGINT NOT NULL,
refcount BIGINT NOT NULL,
PRIMARY KEY (repo_id, oid)
);",
q("git_repos"),
q("git_refs"),
q("registry_seq"),
q("registry_seq"),
q("repo_mutation_markers"),
q("lfs_objects"),
),
tables: &[
"git_repos",
"git_refs",
"registry_seq",
"repo_mutation_markers",
"lfs_objects",
],
}]
}
async fn bootstrap_stamp_table(
client: &mut tokio_postgres::Client,
schema: &str,
) -> Result<(), OpenError> {
let stamp_table = qualify(schema, STAMP_TABLE);
let bootstrap = format!(
"CREATE SCHEMA IF NOT EXISTS {};
CREATE TABLE IF NOT EXISTS {} (
id INTEGER PRIMARY KEY CHECK (id = 1),
stamp BIGINT NOT NULL
);
INSERT INTO {} (id, stamp) VALUES (1, 0) ON CONFLICT (id) DO NOTHING;",
quote_identifier(schema),
stamp_table,
stamp_table,
);
let ctx = || format!("stamp-table bootstrap ({STAMP_TABLE})");
let tx = client
.transaction()
.await
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("{}: BEGIN: {e}", ctx()),
})?;
tx.execute(
"SELECT pg_advisory_xact_lock($1)",
&[&migration_lock_key(schema)],
)
.await
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("{}: {e}", ctx()),
})?;
match tx.batch_execute(&bootstrap).await {
Ok(()) => tx.commit().await.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("{}: commit: {e}", ctx()),
}),
Err(e) => {
drop(tx);
if stamp_table_present(client, schema).await? {
return Ok(());
}
Err(OpenError::Migration {
step: 0,
reason: format!("{}: {e}", ctx()),
})
}
}
}
async fn stamp_table_present(
client: &tokio_postgres::Client,
schema: &str,
) -> Result<bool, OpenError> {
client
.query_one(
"SELECT to_regclass($1) IS NOT NULL",
&[&format!("{schema}.{STAMP_TABLE}")],
)
.await
.map(|row| row.get::<_, bool>(0))
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("stamp-table presence query ({STAMP_TABLE}): {e}"),
})
}
async fn step_effect_present(
client: &tokio_postgres::Client,
schema: &str,
step: &PgStep,
) -> Result<bool, OpenError> {
client
.query_one(
"SELECT (SELECT count(*) FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE n.nspname = $1 AND c.relname = ANY($2)) = $3",
&[&schema, &step.tables, &(step.tables.len() as i64)],
)
.await
.map(|row| row.get::<_, bool>(0))
.map_err(|e| OpenError::Migration {
step: step.version as usize,
reason: format!(
"step {} ({}): verification query: {e}",
step.version, step.name
),
})
}
async fn migrate(client: &mut tokio_postgres::Client, schema: &str) -> Result<i64, OpenError> {
bootstrap_stamp_table(client, schema).await?;
for step in build_steps(schema) {
apply_step(client, schema, &step).await?;
}
let read = format!(
"SELECT stamp FROM {} WHERE id = 1",
qualify(schema, STAMP_TABLE)
);
client
.query_one(&read, &[])
.await
.map(|row| row.get::<_, i64>(0))
.map_err(|e| OpenError::Migration {
step: 0,
reason: format!("stamp read ({STAMP_TABLE}) in {schema}: {e}"),
})
}
async fn apply_step(
client: &mut tokio_postgres::Client,
schema: &str,
step: &PgStep,
) -> Result<(), OpenError> {
let step_no = step.version as usize;
let stamp_read = format!(
"SELECT stamp FROM {} WHERE id = 1",
qualify(schema, STAMP_TABLE)
);
let stamp_advance = format!(
"UPDATE {} SET stamp = $1 WHERE id = 1",
qualify(schema, STAMP_TABLE)
);
let stamp_heal = format!(
"UPDATE {} SET stamp = $1 WHERE id = 1 AND stamp < $1",
qualify(schema, STAMP_TABLE)
);
let step_ctx = || format!("step {step_no} ({})", step.name);
let tx = client
.transaction()
.await
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("{}: BEGIN: {e}", step_ctx()),
})?;
tx.execute(
"SELECT pg_advisory_xact_lock($1)",
&[&migration_lock_key(schema)],
)
.await
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("{}: {e}", step_ctx()),
})?;
let stamp_now: i64 = tx
.query_one(&stamp_read, &[])
.await
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("{}: stamp read: {e}", step_ctx()),
})?
.try_get(0)
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("{}: stamp read: {e}", step_ctx()),
})?;
if stamp_now >= step.version {
tx.commit().await.ok();
return Ok(());
}
let outcome = tx.batch_execute(&step.sql).await;
let outcome = match outcome {
Ok(()) => tx.execute(&stamp_advance, &[&step.version]).await,
Err(e) => Err(e),
};
match outcome {
Ok(_) => tx.commit().await.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("{}: commit: {e}", step_ctx()),
}),
Err(e) => {
drop(tx);
if step_effect_present(client, schema, step).await? {
client
.execute(&stamp_heal, &[&step.version])
.await
.map_err(|e| OpenError::Migration {
step: step_no,
reason: format!("{}: race-swallow stamp heal: {e}", step_ctx()),
})?;
return Ok(());
}
Err(OpenError::Migration {
step: step_no,
reason: format!("{}: {e}", step_ctx()),
})
}
}
}
}
#[cfg(feature = "postgres")]
pub use pg_twin::{CoTenantPg, DEFAULT_SCHEMA};
#[cfg(feature = "postgres")]
pub(crate) use pg_twin::{git_refs_lock_key, lfs_objects_lock_key, qualify};
#[cfg(all(test, feature = "sqlite"))]
mod tests {
use super::sqlite_twin::{
failing_step_probe, read_stamp, run_steps, step_probe, BUSY_TIMEOUT_MS,
STAMP_TABLE_BOOTSTRAP,
};
use super::*;
use crate::registry::{InMemoryRegistry, RegistryRead, RegistryWrite};
use crate::testutil::{mem_registry_pair, unique_db_path};
fn columns_of(conn: &rusqlite::Connection, table: &str) -> Vec<String> {
let mut stmt = conn
.prepare(&format!("PRAGMA table_info({table})"))
.unwrap();
stmt.query_map([], |r| r.get::<_, String>(1))
.unwrap()
.map(|c| c.unwrap())
.collect()
}
fn table_exists(conn: &rusqlite::Connection, table: &str) -> bool {
conn.query_row(
"SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
[table],
|r| r.get::<_, i64>(0),
)
.unwrap()
> 0
}
#[test]
fn fresh_create_stamps_one_and_builds_the_full_co_tenant_schema() {
let path = unique_db_path("migrations", "fresh-create");
let handle = CoTenantSqlite::open(&path).unwrap();
assert_eq!(handle.stamp().unwrap(), 1);
assert_eq!(CoTenantSqlite::step_count(), 1);
let db = handle.lock();
assert_eq!(
db.query_row("PRAGMA busy_timeout", [], |r| r.get::<_, i64>(0))
.unwrap(),
BUSY_TIMEOUT_MS,
"the pinned busy timeout rides the co-tenant connection"
);
assert_eq!(
db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
.unwrap(),
"wal"
);
assert_eq!(
db.query_row("SELECT value FROM registry_seq WHERE id = 1", [], |r| r
.get::<_, i64>(0))
.unwrap(),
0,
"the watermark singleton seeds at 0 on a fresh store"
);
for table in [
"git_repos",
"git_refs",
"registry_seq",
"repo_mutation_markers",
"lfs_objects",
] {
assert!(table_exists(&db, table));
}
assert_eq!(
columns_of(&db, "git_repos"),
vec![
"repo_id".to_string(),
"visibility".to_string(),
"grants".to_string(),
"seq".to_string()
],
"the record rows carry the RepoRecord serde shape's members plus the mutation seq"
);
assert_eq!(
columns_of(&db, "git_refs"),
vec![
"repo_id".to_string(),
"name".to_string(),
"oid_hex".to_string(),
"symref_target".to_string()
],
"HEAD lands as the symref row shape: oid_hex NULL + symref_target set"
);
assert_eq!(
columns_of(&db, "lfs_objects"),
vec![
"repo_id".to_string(),
"oid".to_string(),
"size".to_string(),
"refcount".to_string()
],
"the LFS liveness rows carry the day-one refcount column (ADR-024 §5)"
);
assert_eq!(
columns_of(&db, "repo_mutation_markers"),
vec!["repo_id".to_string(), "seq".to_string()],
);
drop(db);
let reopened = rusqlite::Connection::open(&path).expect("second opener on one file");
assert_eq!(read_stamp(&reopened).unwrap(), 1, "the stamp is durable");
assert!(
table_exists(&reopened, "git_repos"),
"the schema is durable"
);
drop(reopened);
let _ = std::fs::remove_file(&path);
}
#[test]
fn fresh_create_pins_the_visibility_domain() {
let path = unique_db_path("migrations", "visibility-check");
let handle = CoTenantSqlite::open(&path).unwrap();
let db = handle.lock();
assert!(
db.execute(
"INSERT INTO git_repos (repo_id, visibility, grants, seq) \
VALUES ('r', 'weird', '{}', 1)",
[]
)
.is_err(),
"the visibility CHECK pins the record domain at the table"
);
drop(db);
let _ = std::fs::remove_file(&path);
}
#[test]
fn additive_append_applies_step_two_then_is_idempotent() {
let mut db = rusqlite::Connection::open_in_memory().unwrap();
let steps = vec![
step_probe("m_probe_one", "probe one", 1),
step_probe("m_probe_two", "probe two — additive append", 2),
];
let stamp = run_steps(&mut db, &steps).unwrap();
assert_eq!(stamp, 2, "both steps applied and stamped");
assert!(table_exists(&db, "m_probe_one"));
assert!(table_exists(&db, "m_probe_two"));
let again = run_steps(&mut db, &steps).unwrap();
assert_eq!(again, 2, "a re-run is a no-op — steps skip on the stamp");
assert_eq!(
db.query_row(
"SELECT count(*) FROM alkgit_schema_stamp WHERE id = 1",
[],
|r| r.get::<_, i64>(0)
)
.unwrap(),
1,
"one stamp row, always"
);
let appended = vec![step_probe(
"m_probe_three",
"probe three — appended later",
3,
)];
let stamp = run_steps(&mut db, &appended).unwrap();
assert_eq!(stamp, 3, "an additive append advances the stamp one step");
assert!(table_exists(&db, "m_probe_three"));
}
#[test]
fn concurrent_open_race_on_one_file_swallows_the_benign_race() {
let path = unique_db_path("migrations", "concurrent-open");
let stamps = std::thread::scope(|scope| {
let mut joins = Vec::new();
for _ in 0..4 {
joins.push(scope.spawn(|| {
CoTenantSqlite::open(&path)
.map(|handle| handle.stamp().expect("stamped handle reads"))
}));
}
joins
.into_iter()
.map(|join| join.join().expect("no opener thread panics"))
.collect::<Vec<_>>()
});
for stamp in &stamps {
assert_eq!(
stamp.as_ref().expect("every opener succeeds"),
&1,
"the benign race is swallowed — all openers converge on the stamped schema"
);
}
let _ = std::fs::remove_file(&path);
}
#[test]
fn a_pre_applied_effect_swallows_and_heals_the_stamp() {
let mut db = rusqlite::Connection::open_in_memory().unwrap();
db.execute_batch(STAMP_TABLE_BOOTSTRAP).unwrap();
db.execute("CREATE TABLE m_probe_one (x TEXT, y TEXT)", [])
.unwrap();
assert_eq!(
read_stamp(&db).unwrap(),
0,
"the concurrent applier's effect is present, its stamp write not landed"
);
let steps = vec![step_probe("m_probe_one", "probe one", 1)];
run_steps(&mut db, &steps).unwrap();
assert_eq!(
read_stamp(&db).unwrap(),
1,
"the race swallowed on the presence re-key and healed the stamp"
);
assert!(table_exists(&db, "m_probe_one"));
let again = run_steps(&mut db, &steps).unwrap();
assert_eq!(again, 1, "the healed store is idempotent on reopen");
}
#[test]
fn a_failing_step_fails_loudly_and_propagates() {
let path = unique_db_path("migrations", "loud-failure");
let mut db = rusqlite::Connection::open(&path).unwrap();
let steps = vec![
step_probe("m_probe_good", "probe good", 1),
failing_step_probe("m_probe_bad", "probe bad — fails inside its transaction", 2),
step_probe("m_probe_after", "probe after — never reached", 3),
];
let err = run_steps(&mut db, &steps).unwrap_err();
assert!(matches!(err, OpenError::Migration { step: 2, .. }));
match &err {
OpenError::Migration { reason, .. } => {
assert!(reason.contains("already exists"), "{reason}");
}
other => panic!("unexpected variant: {other:?}"),
}
assert_eq!(
read_stamp(&db).unwrap(),
1,
"step 1 committed, step 2 did not"
);
assert!(table_exists(&db, "m_probe_good"));
assert!(
!table_exists(&db, "m_probe_bad"),
"the failed tx rolled back whole"
);
assert!(!table_exists(&db, "m_probe_after"), "later steps never ran");
drop(db);
let mut retry = rusqlite::Connection::open(&path).unwrap();
let retry_err = run_steps(&mut retry, &steps).expect_err("the run fails again");
assert!(
matches!(retry_err, OpenError::Migration { step: 2, .. }),
"the half-migrated store refuses service again — loud on every construction"
);
drop(retry);
let _ = std::fs::remove_file(&path);
}
#[test]
fn an_unknown_forward_stamp_is_safe_to_open_against() {
let path = unique_db_path("migrations", "forward-stamp");
let handle = CoTenantSqlite::open(&path).unwrap();
{
let db = handle.lock();
db.execute("UPDATE alkgit_schema_stamp SET stamp = 99 WHERE id = 1", [])
.unwrap();
}
let reopened = CoTenantSqlite::open(&path).unwrap();
assert_eq!(
reopened.stamp().unwrap(),
99,
"no refuse-gate ships in v1 — the forward stamp is left untouched"
);
assert!(
table_exists(&reopened.lock(), "git_repos"),
"the schema is intact"
);
let mut db = rusqlite::Connection::open_in_memory().unwrap();
let steps = vec![step_probe("m_probe_one", "probe one", 1)];
run_steps(&mut db, &steps).unwrap();
db.execute("UPDATE alkgit_schema_stamp SET stamp = 42 WHERE id = 1", [])
.unwrap();
let forward_steps = vec![
step_probe("m_probe_two", "probe two", 2),
step_probe("m_probe_three", "probe three", 3),
];
let stamp = run_steps(&mut db, &forward_steps).unwrap();
assert_eq!(stamp, 42, "pending steps skip on the forward stamp");
assert!(
!table_exists(&db, "m_probe_two"),
"nothing applied ahead of the stamp"
);
drop(handle);
let _ = std::fs::remove_file(&path);
}
#[test]
fn memory_mode_is_rejected_at_the_handle() {
assert!(
matches!(CoTenantSqlite::open(":memory:"), Err(OpenError::Invalid(_))),
":memory: confers no co-tenancy — the handle rejects the shape (DQ-7)"
);
}
#[tokio::test]
async fn in_memory_adapter_is_stamp_free_fresh_on_every_construction() {
let record = alkgit::RepoRecord {
repo_id: "a".to_string(),
visibility: alkgit::Visibility::Public,
grants: Default::default(),
};
let first = InMemoryRegistry::new();
assert_eq!(first.put(&record).await.unwrap(), 1);
let second = InMemoryRegistry::new();
assert_eq!(
second.resolve("a"),
None,
"a fresh construction builds the current state empty — nothing durable is versioned"
);
assert_eq!(second.watermark().await.unwrap(), 0);
let (_, store_one) = mem_registry_pair();
let (_, store_two) = mem_registry_pair();
store_one.put_repo(&record).await.unwrap();
assert_eq!(
store_two.resolve("a").await.unwrap_err(),
alkgit::RegistryError::NotFound,
"two in-memory constructions are independent stores, not two views of one"
);
}
}