src/server/leases.rs
Ref: Size: 22.7 KiB History
//! Work leases on issues: the one collaboration primitive git cannot
//! express (atomic claim with TTL). One SQLite database per server,
//! `(repo, issue_id)` primary key, lazy expiry. See
//! docs/superpowers/specs/2026-09-05-server-authoritative-collab-design.md.
//!
//! **A row is a tenure ledger for one issue, and is never deleted.** Freeing
//! a lease — by release or by expiry — clears `holder`, leaving `token`
//! behind as a high-water mark that only ever increments. Deleting the row
//! instead (an earlier draft did, and this plan's Task 1 told it to) resets
//! the counter, so tenure 1's zombie could later present token 1 to a fresh
//! tenure 1 and pass a fencing check it should fail. Monotonicity per issue
//! is the whole property a fencing token has, so nothing may reclaim these
//! rows: not release, and not the expiry sweep that `current`/`list` were
//! doing on every read.
use std::path::Path;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior};
/// A live or expired lease row.
#[derive(Debug, Clone, PartialEq)]
pub struct Lease {
pub repo: String,
pub issue_id: String,
pub holder: String,
pub token: i64,
pub acquired_at: i64,
/// None = open-ended (human assignment).
pub expires_at: Option<i64>,
}
/// A row as stored: `holder: None` is a free issue that has been leased
/// before, and whose `token` must outlive the tenure that freed it.
///
/// Liveness is defined once, here, by `held` — an earlier draft also had a
/// `Lease::live`, and two rules for the same question is how they drift.
#[derive(Debug, Clone)]
struct Row {
holder: Option<String>,
token: i64,
acquired_at: i64,
expires_at: Option<i64>,
}
impl Row {
/// Held right now: someone owns it and the clock has not run out.
fn held(&self, now: i64) -> bool {
self.holder.is_some() && self.expires_at.map(|e| e > now).unwrap_or(true)
}
/// The public form, for a row known to be held.
fn into_lease(self, repo: &str, issue_id: &str) -> Option<Lease> {
Some(Lease {
repo: repo.to_string(),
issue_id: issue_id.to_string(),
holder: self.holder?,
token: self.token,
acquired_at: self.acquired_at,
expires_at: self.expires_at,
})
}
}
#[derive(Debug, PartialEq)]
pub enum Acquire {
/// Caller now holds the lease (fresh tenure, or idempotent re-acquire).
Acquired { token: i64, expires_at: Option<i64> },
/// A different holder has a live lease.
Held {
holder: String,
expires_at: Option<i64>,
},
}
#[derive(Debug, PartialEq)]
pub enum Renew {
Renewed {
token: i64,
expires_at: Option<i64>,
},
/// No live lease held by caller (expired, released, or someone else's).
NotHolder {
holder: Option<String>,
},
}
#[derive(Debug, PartialEq)]
pub enum Release {
/// The lease was held and is now free.
Released,
/// There was nothing to release. Still a success — a client retrying after
/// a dropped connection must not fail — but reported distinctly, because
/// answering "released" to a release that freed nothing is what hid the
/// prefix bug of 2026-09-06 for a whole demo.
NotHeld,
/// A different holder has a live lease; refuse.
NotHolder { holder: String },
}
/// The `repo` column value for a repository on disk: its path relative to
/// `repos_dir` (so `…/repos/myrepo.git` becomes `myrepo.git`).
///
/// One function because two callers need the identical string and they reach
/// it from different directions — the SSH verb from a resolved exec path, the
/// web UI from a `RepoEntry`. Deriving it twice is how they would drift, and
/// the symptom would be a claim that exists over SSH but is invisible on the
/// page. Relative rather than absolute so the rows survive `repos_dir`
/// moving.
pub fn repo_key(repos_dir: &Path, repo_path: &Path) -> String {
repo_path
.strip_prefix(repos_dir)
.unwrap_or(repo_path)
.to_string_lossy()
.to_string()
}
/// What an id or prefix names among a repository's lease rows.
#[derive(Debug, PartialEq)]
pub enum Resolved {
/// Exactly one row. Its full `issue_id`.
One(String),
/// No row — nothing is leased under that id.
None,
/// Several rows share the prefix; the caller must be more specific.
Ambiguous(usize),
}
/// Resolve an issue id or prefix against the rows this repository actually
/// holds leases for.
///
/// `acquire` resolves a prefix against the repository's issue refs, because it
/// has to know the issue exists and is open. `renew` and `release` cannot: the
/// issue may have been closed, or its ref deleted, while a lease is still
/// held, and a lease must stay releasable either way. So they resolve against
/// the lease table, which is the only thing that can answer "which lease did
/// you mean".
///
/// Without this, releasing by prefix silently touched a key no row used and
/// reported success, leaving the lease held until its TTL ran out — found by
/// running two real workers against a real server on 2026-09-06, not by any
/// test, because every test used a full id on both sides.
pub fn resolve(conn: &Connection, repo: &str, id_or_prefix: &str) -> rusqlite::Result<Resolved> {
// An exact hit wins without a scan, and cannot be ambiguous.
let exact: Option<String> = conn
.query_row(
"SELECT issue_id FROM leases WHERE repo = ?1 AND issue_id = ?2",
(repo, id_or_prefix),
|row| row.get(0),
)
.optional()?;
if let Some(id) = exact {
return Ok(Resolved::One(id));
}
let mut stmt = conn.prepare(
"SELECT issue_id FROM leases WHERE repo = ?1 AND issue_id LIKE ?2 || '%' LIMIT 2",
)?;
let matches: Vec<String> = stmt
.query_map((repo, id_or_prefix), |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?;
Ok(match matches.len() {
0 => Resolved::None,
1 => Resolved::One(matches.into_iter().next().unwrap_or_default()),
n => Resolved::Ambiguous(n),
})
}
/// Open (creating if needed) the lease database and ensure the schema.
pub fn open(path: &Path) -> rusqlite::Result<Connection> {
let conn = Connection::open(path)?;
init(&conn)?;
Ok(conn)
}
/// Set pragmas and ensure the schema on an already-open connection.
/// Factored out of `open` so tests can run against an in-memory database.
fn init(conn: &Connection) -> rusqlite::Result<()> {
// WAL so a web-UI read never blocks behind an SSH-session write;
// busy_timeout so two SSH sessions serialize instead of erroring.
conn.pragma_update(None, "journal_mode", "WAL")?;
conn.pragma_update(None, "busy_timeout", 5000)?;
// `holder` is nullable because a freed row keeps its token; see the
// module docs on why rows are never deleted.
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS leases (
repo TEXT NOT NULL,
issue_id TEXT NOT NULL,
holder TEXT,
token INTEGER NOT NULL,
acquired_at INTEGER NOT NULL,
expires_at INTEGER,
PRIMARY KEY (repo, issue_id)
)",
)
}
fn get_row(conn: &Connection, repo: &str, issue_id: &str) -> rusqlite::Result<Option<Row>> {
conn.query_row(
"SELECT holder, token, acquired_at, expires_at
FROM leases WHERE repo = ?1 AND issue_id = ?2",
(repo, issue_id),
|row| {
Ok(Row {
holder: row.get(0)?,
token: row.get(1)?,
acquired_at: row.get(2)?,
expires_at: row.get(3)?,
})
},
)
.optional()
}
pub fn acquire(
conn: &mut Connection,
repo: &str,
issue_id: &str,
holder: &str,
ttl_secs: Option<i64>,
now: i64,
) -> rusqlite::Result<Acquire> {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let existing = get_row(&tx, repo, issue_id)?;
let expires_at = ttl_secs.map(|t| now + t);
let outcome = match existing {
Some(row) if row.held(now) => {
if row.holder.as_deref() == Some(holder) {
// Idempotent re-acquire: same tenure, same token, fresh expiry
// from THIS call's ttl. A retrying client must not deadlock
// against itself.
tx.execute(
"UPDATE leases SET expires_at = ?3 WHERE repo = ?1 AND issue_id = ?2",
(repo, issue_id, expires_at),
)?;
Acquire::Acquired {
token: row.token,
expires_at,
}
} else {
Acquire::Held {
// held(now) proved this is Some.
holder: row.holder.unwrap_or_default(),
expires_at: row.expires_at,
}
}
}
other => {
// Free, released, or expired: take a new tenure. The token
// increments off whatever the row remembers — never off nothing,
// because the row is never deleted — and is the fencing token
// later phases will enforce.
let token = other.map(|r| r.token + 1).unwrap_or(1);
tx.execute(
"INSERT OR REPLACE INTO leases
(repo, issue_id, holder, token, acquired_at, expires_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
(repo, issue_id, holder, token, now, expires_at),
)?;
Acquire::Acquired { token, expires_at }
}
};
tx.commit()?;
Ok(outcome)
}
pub fn renew(
conn: &mut Connection,
repo: &str,
issue_id: &str,
holder: &str,
ttl_secs: Option<i64>,
now: i64,
) -> rusqlite::Result<Renew> {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let existing = get_row(&tx, repo, issue_id)?;
let outcome = match existing {
Some(row) if row.held(now) && row.holder.as_deref() == Some(holder) => {
let expires_at = ttl_secs.map(|t| now + t);
tx.execute(
"UPDATE leases SET expires_at = ?3 WHERE repo = ?1 AND issue_id = ?2",
(repo, issue_id, expires_at),
)?;
Renew::Renewed {
token: row.token,
expires_at,
}
}
Some(row) if row.held(now) => Renew::NotHolder { holder: row.holder },
_ => Renew::NotHolder { holder: None },
};
tx.commit()?;
Ok(outcome)
}
pub fn release(
conn: &mut Connection,
repo: &str,
issue_id: &str,
holder: &str,
now: i64,
) -> rusqlite::Result<Release> {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let existing = get_row(&tx, repo, issue_id)?;
let outcome = match existing {
Some(row) if row.held(now) && row.holder.as_deref() != Some(holder) => Release::NotHolder {
holder: row.holder.unwrap_or_default(),
},
Some(row) if row.holder.is_none() => Release::NotHeld,
Some(_) => {
// Free the row, keep the token. Deleting would reset the tenure
// counter — see the module docs.
tx.execute(
"UPDATE leases SET holder = NULL, expires_at = NULL
WHERE repo = ?1 AND issue_id = ?2",
(repo, issue_id),
)?;
Release::Released
}
None => Release::NotHeld,
};
tx.commit()?;
Ok(outcome)
}
/// The lease held on the issue right now, if any.
///
/// Expiry is a *filter*, not a sweep: an expired row stays, so its token
/// survives to fence the tenure that owned it.
pub fn current(
conn: &Connection,
repo: &str,
issue_id: &str,
now: i64,
) -> rusqlite::Result<Option<Lease>> {
Ok(get_row(conn, repo, issue_id)?
.filter(|row| row.held(now))
.and_then(|row| row.into_lease(repo, issue_id)))
}
/// Every lease held in a repo right now, oldest tenure first.
pub fn list(conn: &Connection, repo: &str, now: i64) -> rusqlite::Result<Vec<Lease>> {
let mut stmt = conn.prepare(
"SELECT issue_id, holder, token, acquired_at, expires_at
FROM leases
WHERE repo = ?1 AND holder IS NOT NULL
AND (expires_at IS NULL OR expires_at > ?2)
ORDER BY acquired_at, issue_id",
)?;
let rows = stmt.query_map((repo, now), |row| {
Ok(Lease {
repo: repo.to_string(),
issue_id: row.get(0)?,
holder: row.get(1)?,
token: row.get(2)?,
acquired_at: row.get(3)?,
expires_at: row.get(4)?,
})
})?;
rows.collect()
}
#[cfg(test)]
mod tests {
use super::*;
fn mem() -> Connection {
let conn = Connection::open_in_memory().unwrap();
init(&conn).unwrap();
conn
}
#[test]
fn acquire_free_issue_returns_token_1() {
let mut c = mem();
let got = acquire(&mut c, "r", "i", "alice", None, 100).unwrap();
assert_eq!(
got,
Acquire::Acquired {
token: 1,
expires_at: None
}
);
}
#[test]
fn acquire_sets_expiry_from_ttl() {
let mut c = mem();
let got = acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
assert_eq!(
got,
Acquire::Acquired {
token: 1,
expires_at: Some(400)
}
);
}
#[test]
fn acquire_without_ttl_is_open_ended() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", None, 100).unwrap();
// Far future: still held.
let lease = current(&c, "r", "i", i64::MAX - 1).unwrap().unwrap();
assert_eq!(lease.holder, "alice");
assert_eq!(lease.expires_at, None);
}
#[test]
fn second_acquire_by_other_holder_returns_held() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
let got = acquire(&mut c, "r", "i", "bob", Some(300), 150).unwrap();
assert_eq!(
got,
Acquire::Held {
holder: "alice".into(),
expires_at: Some(400)
}
);
}
#[test]
fn reacquire_by_holder_is_idempotent_same_token_new_expiry() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
let got = acquire(&mut c, "r", "i", "alice", Some(300), 200).unwrap();
assert_eq!(
got,
Acquire::Acquired {
token: 1,
expires_at: Some(500)
}
);
}
#[test]
fn acquire_after_expiry_takes_over_and_bumps_token() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
let got = acquire(&mut c, "r", "i", "bob", Some(300), 500).unwrap();
assert_eq!(
got,
Acquire::Acquired {
token: 2,
expires_at: Some(800)
}
);
}
#[test]
fn renew_extends_expiry_keeps_token() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
let got = renew(&mut c, "r", "i", "alice", Some(300), 200).unwrap();
assert_eq!(
got,
Renew::Renewed {
token: 1,
expires_at: Some(500)
}
);
}
#[test]
fn renew_by_non_holder_returns_not_holder() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
let got = renew(&mut c, "r", "i", "bob", Some(300), 200).unwrap();
assert_eq!(
got,
Renew::NotHolder {
holder: Some("alice".into())
}
);
}
#[test]
fn renew_after_expiry_returns_not_holder() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
let got = renew(&mut c, "r", "i", "alice", Some(300), 500).unwrap();
assert_eq!(got, Renew::NotHolder { holder: None });
}
#[test]
fn release_by_holder_frees_the_issue() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
assert_eq!(
release(&mut c, "r", "i", "alice", 200).unwrap(),
Release::Released
);
assert_eq!(current(&c, "r", "i", 200).unwrap(), None);
}
#[test]
fn release_of_nothing_succeeds_but_says_so() {
let mut c = mem();
// Still a success for a retrying client, but not called "released":
// reporting a freed lease when nothing was freed hid a real bug.
assert_eq!(
release(&mut c, "r", "i", "alice", 200).unwrap(),
Release::NotHeld
);
}
#[test]
fn resolve_finds_an_exact_id_and_a_unique_prefix() {
let mut c = mem();
acquire(&mut c, "r", "abcdef123456", "alice", None, 100).unwrap();
assert_eq!(
resolve(&c, "r", "abcdef123456").unwrap(),
Resolved::One("abcdef123456".into())
);
assert_eq!(
resolve(&c, "r", "abcd").unwrap(),
Resolved::One("abcdef123456".into())
);
}
#[test]
fn resolve_reports_nothing_and_ambiguity() {
let mut c = mem();
acquire(&mut c, "r", "abcdef", "alice", None, 100).unwrap();
acquire(&mut c, "r", "abcxyz", "bob", None, 100).unwrap();
assert_eq!(resolve(&c, "r", "zz").unwrap(), Resolved::None);
assert_eq!(resolve(&c, "r", "abc").unwrap(), Resolved::Ambiguous(2));
// Another repo's rows are not candidates.
assert_eq!(resolve(&c, "other", "abc").unwrap(), Resolved::None);
}
#[test]
fn resolve_still_finds_a_freed_row_so_a_stale_id_is_answerable() {
let mut c = mem();
acquire(&mut c, "r", "abcdef", "alice", None, 100).unwrap();
release(&mut c, "r", "abcdef", "alice", 200).unwrap();
// The row survives to carry the token, so the id still resolves — and
// releasing it again answers NotHeld rather than inventing a lease.
assert_eq!(
resolve(&c, "r", "abc").unwrap(),
Resolved::One("abcdef".into())
);
}
#[test]
fn release_by_non_holder_refused() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
assert_eq!(
release(&mut c, "r", "i", "bob", 200).unwrap(),
Release::NotHolder {
holder: "alice".into()
}
);
}
#[test]
fn release_of_expired_lease_by_anyone_succeeds() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
assert_eq!(
release(&mut c, "r", "i", "bob", 500).unwrap(),
Release::Released
);
}
#[test]
fn list_shows_only_live_leases() {
let mut c = mem();
acquire(&mut c, "r", "a", "alice", Some(300), 100).unwrap();
acquire(&mut c, "r", "b", "bob", Some(100), 100).unwrap();
acquire(&mut c, "r", "c", "carol", None, 150).unwrap();
acquire(&mut c, "other", "a", "dave", None, 100).unwrap();
let live = list(&c, "r", 250).unwrap();
let holders: Vec<&str> = live.iter().map(|l| l.holder.as_str()).collect();
assert_eq!(holders, vec!["alice", "carol"]);
}
#[test]
fn current_none_after_expiry() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
assert!(current(&c, "r", "i", 200).unwrap().is_some());
assert_eq!(current(&c, "r", "i", 400).unwrap(), None);
}
#[test]
fn token_does_not_reset_after_release() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
release(&mut c, "r", "i", "alice", 200).unwrap();
// A released issue is free, but its tenure history is not forgotten:
// the next holder must not be handed alice's token back.
let got = acquire(&mut c, "r", "i", "bob", Some(300), 300).unwrap();
assert_eq!(
got,
Acquire::Acquired {
token: 2,
expires_at: Some(600)
}
);
}
#[test]
fn token_does_not_reset_after_expiry_and_reads() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(100), 0).unwrap();
// Reads once swept expired rows, which reset the counter. They must
// filter instead.
assert_eq!(current(&c, "r", "i", 500).unwrap(), None);
assert_eq!(list(&c, "r", 500).unwrap().len(), 0);
let got = acquire(&mut c, "r", "i", "bob", Some(100), 500).unwrap();
assert_eq!(
got,
Acquire::Acquired {
token: 2,
expires_at: Some(600)
}
);
}
#[test]
fn released_issue_is_free_to_anyone() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(300), 100).unwrap();
release(&mut c, "r", "i", "alice", 200).unwrap();
assert_eq!(current(&c, "r", "i", 200).unwrap(), None);
assert!(matches!(
acquire(&mut c, "r", "i", "bob", None, 200).unwrap(),
Acquire::Acquired { .. }
));
}
#[test]
fn tenure_token_monotonic_across_holders() {
let mut c = mem();
acquire(&mut c, "r", "i", "alice", Some(100), 0).unwrap();
let b = acquire(&mut c, "r", "i", "bob", Some(100), 200).unwrap();
assert_eq!(
b,
Acquire::Acquired {
token: 2,
expires_at: Some(300)
}
);
let a = acquire(&mut c, "r", "i", "alice", Some(100), 400).unwrap();
assert_eq!(
a,
Acquire::Acquired {
token: 3,
expires_at: Some(500)
}
);
}
#[test]
fn repo_key_is_relative_to_repos_dir() {
assert_eq!(
repo_key(Path::new("/srv/git"), Path::new("/srv/git/myrepo.git")),
"myrepo.git"
);
assert_eq!(
repo_key(Path::new("/srv/git"), Path::new("/srv/git/org/sub.git")),
"org/sub.git"
);
// A path outside repos_dir cannot be made relative; keep it whole
// rather than inventing a key that could collide.
assert_eq!(
repo_key(Path::new("/srv/git"), Path::new("/elsewhere/x.git")),
"/elsewhere/x.git"
);
}
#[test]
fn open_creates_file_and_schema() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("collab.db");
{
let mut conn = open(&path).unwrap();
acquire(&mut conn, "r", "i", "alice", None, 100).unwrap();
}
let conn = open(&path).unwrap();
assert!(current(&conn, "r", "i", 200).unwrap().is_some());
}
}