~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

The index reads through connections of its own, and forgets accounts narrowly

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/index-readers
crates/knotless-serve/src/index.rs+261−82
View file
@@ -22,9 +22,9 @@ use rusqlite::{Connection, OpenFlags, OptionalExtension, params};
use serde::Serialize;
use serde_json::Value;
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::path::Path;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, mpsc};
@@ -45,8 +45,119 @@ fn chosen(choice: &Value) -> Option<(String, Option<Value>)> {
theme.map(|t| (t, palette))
}
/// The most connections reading at once: past it a read waits for one.
const READERS: usize = 8;
/// Connections to the index that only read, each opened as reads first
/// need it, and kept for the next read once one is done with it. Opened
/// as needed, not all at once, as a process with many indexes open, as
/// tests are, would run out of file descriptors.
struct Readers {
path: PathBuf,
/// Those not reading now.
free: Mutex<Vec<Connection>>,
/// How many are open, reading or not.
open: Mutex<usize>,
returned: Condvar,
}
impl Readers {
fn new(path: &Path) -> Readers {
Readers {
path: path.to_path_buf(),
free: Mutex::default(),
open: Mutex::default(),
returned: Condvar::new(),
}
}
/// A connection to read with, given back as it is dropped: a free one,
/// else one opened, else, past [`READERS`], once one is free. One that
/// cannot be opened reads through `writer` instead.
fn take<'a>(&'a self, writer: &'a Mutex<Connection>) -> Reader<'a> {
let mut open = self.open.lock().unwrap();
loop {
if let Some(db) = self.free.lock().unwrap().pop() {
return Reader::own(db, self);
}
if *open < READERS {
let Ok(db) = reader(&self.path) else {
let held = Held::Writer(writer.lock().unwrap());
return Reader {
held: Some(held),
from: self,
};
};
*open += 1;
return Reader::own(db, self);
}
open = self.returned.wait(open).unwrap();
}
}
}
/// A connection the index's tables are read with, as [`connect`] made
/// them: it cannot write.
fn reader(path: &Path) -> rusqlite::Result<Connection> {
let db = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?;
let flags = FunctionFlags::SQLITE_UTF8 | FunctionFlags::SQLITE_DETERMINISTIC;
db.create_scalar_function("zone_day", 2, flags, zone_day)?;
db.busy_timeout(Duration::from_secs(5))?;
db.execute_batch("PRAGMA query_only = ON")?;
crate::telemetry::watch(&db);
Ok(db)
}
/// A connection taken from [`Readers`], given back as it is dropped.
struct Reader<'a> {
held: Option<Held<'a>>,
from: &'a Readers,
}
/// What a [`Reader`] reads through: a connection of its own, or the
/// writer's, when one of its own could not be opened.
enum Held<'a> {
Own(Connection),
Writer(std::sync::MutexGuard<'a, Connection>),
}
impl<'a> Reader<'a> {
fn own(db: Connection, from: &'a Readers) -> Reader<'a> {
let held = Some(Held::Own(db));
Reader { held, from }
}
}
impl std::ops::Deref for Reader<'_> {
type Target = Connection;
fn deref(&self) -> &Connection {
let held = self.held.as_ref();
match held.expect("a reader holds its connection until dropped") {
Held::Own(db) => db,
Held::Writer(db) => db,
}
}
}
impl Drop for Reader<'_> {
fn drop(&mut self) {
if let Some(Held::Own(db)) = self.held.take() {
// Held while it goes back, so a read waiting sees it there.
let open = self.from.open.lock().unwrap();
self.from.free.lock().unwrap().push(db);
drop(open);
self.from.returned.notify_one();
}
}
}
pub struct Index {
/// The one connection that writes, and reads what a write needs.
db: Mutex<Connection>,
/// The connections that only read: a page waits on no write, nor on
/// another page's read, as the database is in WAL mode.
readers: Readers,
/// Whether opening it built its cache anew, empty: a new file, or one
/// of another version of the tables.
pub rebuilt: bool,
@@ -420,6 +531,7 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
started_at TEXT NOT NULL, value TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS check_repo ON check_run (repo_did, repo_key, started_at);
CREATE INDEX IF NOT EXISTS check_sha ON check_run (repo_did, repo_key, sha);
CREATE INDEX IF NOT EXISTS check_author ON check_run (author);
CREATE TABLE IF NOT EXISTS minutes (
uri TEXT PRIMARY KEY, runner TEXT NOT NULL, owner TEXT NOT NULL, value TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS minutes_owner ON minutes (owner);
@@ -715,8 +827,19 @@ const FORGET_REPO: &[&str] = &[
"DELETE FROM language WHERE did = ?1 AND key = ?2",
];
/// What [`Index::purge`] runs, with `?1` the account's DID and `?2` the
/// start of its records' URIs: first the records of the spaces it owned,
/// The repositories, as `<did>/<key>`, whose pulls and issues
/// [`Index::purge`] changes of the account `?1`: its own, those it moved
/// from and to, and those it wrote about or checked.
const PURGED_SCOPES: &[&str] = &[
"SELECT did || '/' || key FROM repo WHERE did = ?1",
"SELECT from_did || '/' || from_key FROM moved WHERE from_did = ?1",
"SELECT to_did || '/' || to_key FROM moved WHERE from_did = ?1",
"SELECT DISTINCT repo_did || '/' || repo_key FROM forge WHERE author = ?1",
"SELECT DISTINCT repo_did || '/' || repo_key FROM check_run WHERE author = ?1",
];
/// What [`Index::purge`] runs, with `?1` the account's DID, `?2` the
/// start of its records' URIs and `?3` the first URI past them: first the records of the spaces it owned,
/// whoever wrote them, then each table of what it wrote or was kept for it.
const PURGE: &[&str] = &[
"CREATE TEMP TABLE IF NOT EXISTS purged (uri TEXT PRIMARY KEY)",
@@ -729,7 +852,7 @@ const PURGE: &[&str] = &[
OR uri IN (SELECT uri FROM forge WHERE author = ?1)",
"DELETE FROM forge WHERE author = ?1 OR uri IN purged",
"DELETE FROM comment WHERE author = ?1 OR uri IN purged",
"DELETE FROM mention WHERE did = ?1 OR substr(uri, 1, length(?2)) = ?2 OR uri IN purged",
"DELETE FROM mention WHERE did = ?1 OR (uri >= ?2 AND uri < ?3) OR uri IN purged",
"DELETE FROM check_run WHERE author = ?1 OR uri IN purged",
"DELETE FROM minutes WHERE runner = ?1 OR owner = ?1 -- scans: minutes, once, as an account is forgotten",
"DELETE FROM runner_key WHERE did = ?1",
@@ -783,6 +906,7 @@ impl Index {
let (read, reading) = mpsc::unbounded_channel();
let index = Index {
db: Mutex::new(db),
readers: Readers::new(path),
rebuilt,
changes: broadcast::channel(1024).0,
asked: Mutex::default(),
@@ -878,7 +1002,7 @@ impl Index {
/// how many have failed since, the last of them asking for no other
/// before `retry_at`, in seconds since the epoch.
pub fn backfilled(&self, did: &str) -> rusqlite::Result<Backfilled> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let found = db.query_row(
"SELECT rev, tries, retry_at FROM backfill WHERE did = ?1",
[did],
@@ -1252,7 +1376,7 @@ impl Index {
/// What `did`'s repository with `key` is mostly written in, as JSON
/// (`null` for nothing counted), if it was ever counted.
pub fn language(&self, did: &str, key: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt =
db.prepare_cached("SELECT value FROM language WHERE did = ?1 AND key = ?2")?;
stmt.query_row([did, key], |row| row.get(0)).optional()
@@ -1336,7 +1460,7 @@ impl Index {
/// of those that publish one.
pub fn runner_keys(&self, runners: &[String]) -> rusqlite::Result<Vec<(String, String)>> {
let runners = serde_json::to_string(runners).unwrap_or_default();
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT k.did, k.key FROM json_each(?1) r JOIN runner_key k ON k.did = r.value
ORDER BY r.key",
@@ -1350,7 +1474,7 @@ impl Index {
/// a runner taken down.
pub fn minutes(&self, owner: &str, runners: &[String]) -> rusqlite::Result<Vec<Entry>> {
let runners = serde_json::to_string(runners).unwrap_or_default();
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT m.uri, m.runner, m.value FROM json_each(?2) r
JOIN minutes m ON m.runner = r.value AND m.owner = ?1
@@ -1364,7 +1488,7 @@ impl Index {
/// The repository records `did` holds, by record key, as the index
/// has them, each with its URI.
pub fn actor_repos(&self, did: &str) -> rusqlite::Result<Vec<(String, RepoRecord, String)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT name, value, uri FROM repo WHERE did = ?1")?;
let rows = stmt.query_map([did], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?;
let rows: Vec<(String, String, String)> = rows.collect::<rusqlite::Result<_>>()?;
@@ -1378,7 +1502,7 @@ impl Index {
/// The check `uri`, as the index has it.
pub fn check_record(&self, uri: &str) -> rusqlite::Result<Option<Value>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT value FROM check_run WHERE uri = ?1")?;
let found: Option<String> = stmt.query_row([uri], |row| row.get(0)).optional()?;
Ok(found.and_then(|value| serde_json::from_str(&value).ok()))
@@ -1389,7 +1513,7 @@ impl Index {
/// they are read all at once, as its latest of each package is found
/// among them all.
pub fn releases(&self, did: &str, key: &str) -> rusqlite::Result<Vec<(String, ReleaseRecord)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare(
"SELECT uri, value FROM release
WHERE did = ?1 AND repo_key = ?2 AND uri NOT IN down
@@ -1402,7 +1526,7 @@ impl Index {
/// How many accounts starred `did`'s repository with `key`, leaving
/// out those taken down.
pub fn star_count(&self, did: &str, key: &str) -> rusqlite::Result<u64> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT COUNT(DISTINCT did) FROM star
WHERE repo_did = ?1 AND repo_key = ?2 AND did NOT IN down",
@@ -1413,7 +1537,7 @@ impl Index {
/// `viewer`'s star naming `did`'s repository with `key`, by its URI,
/// if they starred it.
pub fn star_of(&self, viewer: &str, did: &str, key: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT uri FROM star WHERE did = ?1 AND repo_did = ?2 AND repo_key = ?3
ORDER BY uri LIMIT 1",
@@ -1425,7 +1549,7 @@ impl Index {
/// The repositories `did` starred that the index knows, as
/// [`Index::repos`] lists them, the latest starred first.
pub fn actor_stars(&self, did: &str) -> rusqlite::Result<Vec<Listed>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT r.did, r.name, r.description, r.fork_of, r.created_at, r.uri
FROM star s JOIN repo r ON r.did = s.repo_did AND r.key = s.repo_key
@@ -1438,7 +1562,7 @@ impl Index {
/// How many accounts follow `did`, and how many it follows, each once
/// however many of its follows name them, leaving out those taken down.
pub fn follow_counts(&self, did: &str) -> rusqlite::Result<(u64, u64)> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT
(SELECT COUNT(DISTINCT did) FROM follow
@@ -1454,7 +1578,7 @@ impl Index {
/// has, each star once per account, and how many repositories it has,
/// leaving out those taken down.
pub fn badge_counts(&self, did: &str) -> rusqlite::Result<(u64, u64, u64)> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT
(SELECT COUNT(*) FROM release WHERE did = ?1 AND uri NOT IN down),
@@ -1484,7 +1608,7 @@ impl Index {
/// When `did` first commented on each pull or issue it commented on,
/// by its URI.
pub fn first_comments(&self, did: &str) -> rusqlite::Result<HashMap<String, String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT subject, MIN(created_at) FROM comment
WHERE author = ?1 AND uri NOT IN down GROUP BY subject",
@@ -1510,7 +1634,7 @@ impl Index {
/// merge each says. A repository the index does not have, or that was
/// taken down, is left out.
fn pulls_where(&self, which: &str, did: &str) -> rusqlite::Result<Vec<Pulls>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
// The pulls, then what is about them by `forge_subject`, in this
// order (`CROSS JOIN`): a join on either, as one `OR`, reads every
// record of the repository for each pull.
@@ -1565,7 +1689,7 @@ impl Index {
key: &str,
batch: &str,
) -> rusqlite::Result<Vec<Entry>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT uri, author, value FROM forge
WHERE json_extract(value, '$.batch') = ?3
@@ -1579,7 +1703,7 @@ impl Index {
/// seconds: whether it moved from one PDS to another, and whether its
/// PDS is one of its own.
pub fn home(&self, did: &str) -> rusqlite::Result<Option<(Home, i64)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT moved, own, checked FROM home WHERE did = ?1")?;
stmt.query_row([did], |row| {
let home = Home {
@@ -1617,7 +1741,7 @@ impl Index {
viewer: Option<&str>,
) -> rusqlite::Result<Vec<Reacted>> {
let subjects = serde_json::to_string(subjects).unwrap_or_default();
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT x.subject, x.content, COUNT(DISTINCT x.did),
MIN(CASE WHEN x.did = ?2 THEN x.uri END)
@@ -1640,7 +1764,7 @@ impl Index {
/// `viewer`'s follow naming `did`, by its URI, if they follow it.
pub fn follow_of(&self, viewer: &str, did: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT uri FROM follow WHERE did = ?1 AND subject = ?2 ORDER BY uri LIMIT 1",
)?;
@@ -1650,7 +1774,7 @@ impl Index {
/// `viewer`'s block keeping `did` out of all its repositories, by its
/// URI, if it has one.
pub fn block_of(&self, viewer: &str, did: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT uri FROM block WHERE did = ?1 AND subject = ?2 ORDER BY uri LIMIT 1",
)?;
@@ -1675,7 +1799,7 @@ impl Index {
("subject", "did")
};
let (at, uri) = after.and_then(|a| a.split_once(' ')).unwrap_or(("~", ""));
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(&format!(
"SELECT {who}, MAX(created_at) AS at, MAX(uri) AS u FROM follow
WHERE {whom} = ?1 AND {who} NOT IN down GROUP BY {who}
@@ -1744,7 +1868,7 @@ impl Index {
/// `open`, `closed` or `merged`, by the state records of those who may
/// set it. None when the index has not it, or its repository.
fn item_state(&self, uri: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT r.did, r.value FROM forge p
JOIN repo r ON r.did = p.repo_did AND r.key = p.repo_key WHERE p.uri = ?1",
@@ -1777,7 +1901,7 @@ impl Index {
) -> rusqlite::Result<(Vec<Notified>, Option<String>)> {
let (at, uri) = after.unwrap_or("~ ").split_once(' ').unwrap_or(("~", ""));
let orgs = serde_json::to_string(orgs).unwrap_or_default();
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(&format!(
"WITH
-- The repositories the account looks after, by owner and, for
@@ -2006,7 +2130,7 @@ impl Index {
/// this one keeps: not of accounts it blocked or that are taken down,
/// nor for reasons it would rather not hear of.
pub fn heard(&self, did: &str, found: Vec<Notified>) -> rusqlite::Result<Vec<Notified>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT ?2 NOT IN down
AND ?2 NOT IN (SELECT subject FROM block WHERE did = ?1)
@@ -2027,7 +2151,7 @@ impl Index {
/// When `did` last saw its notifications, if ever.
pub fn seen(&self, did: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT at FROM seen WHERE did = ?1")?;
stmt.query_row([did], |row| row.get(0)).optional()
}
@@ -2049,7 +2173,7 @@ impl Index {
/// The reasons `did` hears no notifications of, and whether it hears
/// only of what accounts it follows did: none and no, until it says.
pub fn notification_preferences(&self, did: &str) -> rusqlite::Result<(Vec<String>, bool)> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT value ->> '$.off', value ->> '$.followedOnly' FROM pref
WHERE did = ?1 AND kind = ?2",
@@ -2075,7 +2199,7 @@ impl Index {
/// `did`'s preference of `kind`, if it set one.
pub fn preference(&self, did: &str, kind: &str) -> rusqlite::Result<Option<Value>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT value FROM pref WHERE did = ?1 AND kind = ?2")?;
let kept: Option<String> = stmt.query_row([did, kind], |row| row.get(0)).optional()?;
Ok(kept.and_then(|v| serde_json::from_str(&v).ok()))
@@ -2098,7 +2222,7 @@ impl Index {
/// At most `limit` themes whose name starts with `prefix`, in any case,
/// in the order of their names, as `(uri, did, record)`.
pub fn themes_named(&self, prefix: &str, limit: u32) -> rusqlite::Result<Vec<Theme>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
// The names from the prefix up to the last that could start so.
let mut stmt = db.prepare_cached(
"SELECT uri, did, created_at, value FROM theme
@@ -2123,7 +2247,7 @@ impl Index {
after: Option<&str>,
) -> rusqlite::Result<(Vec<Theme>, Option<String>)> {
let (at, uri) = after.and_then(|a| a.split_once(' ')).unwrap_or(("~", ""));
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
// Each way in by an index of its own: one account's, or everyone's.
let sql = match actor {
Some(_) => {
@@ -2155,7 +2279,7 @@ impl Index {
/// The themes `uris` names that the index has, of accounts not taken
/// down, in the order named, as `(uri, did, record)`.
pub fn themes_at(&self, uris: &[&str]) -> rusqlite::Result<Vec<Theme>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let named = serde_json::to_string(uris).unwrap_or_default();
let mut stmt = db.prepare_cached(
"SELECT t.uri, t.did, t.created_at, t.value FROM json_each(?1) AS n
@@ -2182,7 +2306,7 @@ impl Index {
/// Whether `did`'s profile is kept: its records, as the firehose
/// changes them.
pub fn profile_kept(&self, did: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT 1 FROM profile_kept WHERE did = ?1")?;
stmt.exists([did])
}
@@ -2193,7 +2317,7 @@ impl Index {
if !self.profile_kept(did)? {
return Ok(None);
}
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT collection, rkey, value FROM profile_record WHERE did = ?1
ORDER BY collection, rkey",
@@ -2226,7 +2350,7 @@ impl Index {
/// taken down, not by an account taken down, and not kept out by the
/// repository's owner.
pub fn forge(&self, did: &str, key: &str) -> rusqlite::Result<Vec<Entry>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
// With those about the repository it moved from, when both owners
// say so: it names that one, and that one's owner said it moved here.
let mut stmt = db.prepare_cached(
@@ -2301,7 +2425,7 @@ impl Index {
/// The repository `did`'s `key` moved from, as `<did>/<key>`, whose
/// pulls and issues [`Index::forge`] reads with its own.
fn moved_from(&self, did: &str, key: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT from_did || '/' || from_key FROM repo
WHERE did = ?1 AND key = ?2 AND from_did IS NOT NULL AND from_key IS NOT NULL",
@@ -2327,7 +2451,7 @@ impl Index {
let commits = filter
.commits
.map(|c| serde_json::to_string(c).unwrap_or_default());
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT uri, author, value, started_at FROM check_run
WHERE repo_did = ?1 AND repo_key = ?2
@@ -2369,7 +2493,7 @@ impl Index {
/// Whether any of `subjects`, accounts or records, is taken down here
/// or by a labeler this server follows.
pub fn down(&self, subjects: &[&str]) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT 1 FROM down WHERE subject = ?1")?;
for s in subjects {
if stmt.exists([s])? {
@@ -2402,7 +2526,7 @@ impl Index {
/// What `did`'s host last said of it, if it is not active.
pub fn status(&self, did: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT status FROM inactive WHERE did = ?1")?;
stmt.query_row([did], |row| row.get(0)).optional()
}
@@ -2416,14 +2540,32 @@ impl Index {
let mut db = self.db.lock().unwrap();
let tx = db.transaction()?;
let own = format!("at://{did}/");
let args = [did, own.as_str()];
// Past the last URI starting `own`: `/` is followed by `0`.
let past = format!("at://{did}0");
let args = [did, own.as_str(), past.as_str()];
// What it wrote about, read before it goes: only those
// repositories' pulls and issues are made again, not every one's.
let mut scopes = vec![did.to_string()];
for sql in PURGED_SCOPES {
let mut stmt = tx.prepare(sql)?;
let found = stmt.query_map([did], |row| row.get::<_, String>(0))?;
scopes.extend(found.collect::<rusqlite::Result<Vec<_>>>()?);
}
for sql in PURGE {
let mut stmt = tx.prepare(sql)?;
let given = args.iter().take(stmt.parameter_count());
stmt.execute(rusqlite::params_from_iter(given))?;
}
// The records of spaces it owned can be about any repository.
let spaced: bool =
tx.query_row("SELECT EXISTS (SELECT 1 FROM purged)", [], |r| r.get(0))?;
tx.commit()?;
self.stamps.mark_all();
if spaced {
self.stamps.mark_all();
}
for scope in &scopes {
self.stamps.mark(scope);
}
Ok(())
}
@@ -2490,7 +2632,7 @@ impl Index {
/// How a takedown of `subject` named it, if it is taken down.
pub fn taken_down(&self, subject: &str) -> rusqlite::Result<Option<Value>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row(
"SELECT value FROM takedown WHERE subject = ?1",
[subject],
@@ -2501,7 +2643,7 @@ impl Index {
/// The labels this server gave out after `seq`, oldest first.
pub fn emitted(&self, seq: i64) -> rusqlite::Result<Vec<(i64, Value)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt =
db.prepare_cached("SELECT seq, label FROM emitted WHERE seq > ?1 ORDER BY seq")?;
stmt.query_map([seq], numbered)?.collect()
@@ -2510,7 +2652,7 @@ impl Index {
/// Takedowns standing whose subject starts with `prefix`, as labels
/// from `src`.
pub fn labels(&self, src: &str, prefix: &str) -> rusqlite::Result<Vec<Value>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT subject, at FROM takedown WHERE substr(subject, 1, length(?1)) = ?1
ORDER BY subject LIMIT 250",
@@ -2574,7 +2716,7 @@ impl Index {
/// Where the labeler `src`'s stream was last read to.
pub fn labeler_seq(&self, src: &str) -> rusqlite::Result<Option<i64>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row("SELECT seq FROM labeler WHERE src = ?1", [src], |row| {
row.get(0)
})
@@ -2583,7 +2725,7 @@ impl Index {
/// Whether the index answers: a read of one of its tables.
pub fn check(&self) -> rusqlite::Result<i64> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row("SELECT count(*) FROM stream", [], |row| row.get(0))
}
@@ -2598,7 +2740,7 @@ impl Index {
month: &str,
) -> rusqlite::Result<Network> {
let since = days.keys().next().cloned().unwrap_or_default();
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let acts = format!(
"-- scans: forge, check_run, push, for the operator's usage report alone
-- sorts: forge, check_run, push, for the same
@@ -2674,7 +2816,7 @@ impl Index {
/// The member whose session the space `uri` is synced with, if any.
pub fn syncer_of(&self, uri: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row(
"SELECT did FROM space_syncer WHERE uri = ?1",
[uri],
@@ -2686,7 +2828,7 @@ impl Index {
/// The spaces to sync, each with the member whose session it is synced
/// with.
pub fn syncers(&self) -> rusqlite::Result<Vec<(String, String)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT uri, did FROM space_syncer ORDER BY uri")?;
stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
.collect()
@@ -2725,7 +2867,7 @@ impl Index {
/// The records of the private repositories `did` was proven to read
/// in someone else's space, by URI.
pub fn read_of(&self, did: &str) -> rusqlite::Result<HashSet<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT r.space || '/' || r.did || '/' || r.collection || '/' || r.rkey
FROM space_reader s JOIN space_record r ON r.space = s.uri
@@ -2740,7 +2882,7 @@ impl Index {
/// organization's record there names them, or where they wrote, and
/// those they were proven to read before, followed or not.
pub fn maybe_member(&self, did: &str) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT uri FROM space WHERE uri NOT LIKE ?2 AND (
EXISTS (SELECT 1 FROM space_record r JOIN repo ON repo.uri =
@@ -2791,7 +2933,7 @@ impl Index {
/// The spaces [`Index::space_gone`] noted.
pub fn gone_spaces(&self) -> rusqlite::Result<HashSet<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT uri FROM space_gone")?;
stmt.query_map([], |row| row.get(0))?.collect()
}
@@ -2799,7 +2941,7 @@ impl Index {
/// The spaces followed, each with where its sync is, if it has begun:
/// the space revision of the newest write it has taken in.
pub fn spaces(&self) -> rusqlite::Result<Vec<(String, Option<String>)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT uri, cursor FROM space ORDER BY uri")?;
stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
.collect()
@@ -2818,7 +2960,7 @@ impl Index {
/// Whether the space `uri` has been synced at all.
pub fn space_synced(&self, uri: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let found = db.query_row("SELECT cursor FROM space WHERE uri = ?1", [uri], |row| {
row.get::<_, Option<String>>(0)
});
@@ -2832,7 +2974,7 @@ impl Index {
uri: &str,
did: &str,
) -> rusqlite::Result<Option<(String, Vec<u8>)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row(
"SELECT rev, set_hash FROM space_writer WHERE space = ?1 AND did = ?2",
[uri, did],
@@ -2863,7 +3005,7 @@ impl Index {
/// collection and rkey: what a sync that reads it whole takes out if
/// it no longer has them.
pub fn space_records(&self, uri: &str, did: &str) -> rusqlite::Result<Vec<(String, String)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT collection, rkey FROM space_record WHERE space = ?1 AND did = ?2",
)?;
@@ -2938,7 +3080,7 @@ impl Index {
/// public records when empty) name, the first invited first, each with
/// whether they keep a record of joining it there.
pub fn org_roster(&self, org: &str, space: &str) -> rusqlite::Result<Vec<Invitee>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT m.did, m.role, m.invited_at, m.expires_at, m.joined_at,
EXISTS (SELECT 1 FROM membership j
@@ -2952,14 +3094,14 @@ impl Index {
/// Whether `did` is an organization: its public records say it is one.
pub fn is_org(&self, did: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT 1 FROM org WHERE space = '' AND did = ?1")?;
stmt.exists([did])
}
/// The gateway the organization `org`'s public record names.
pub fn org_gateway(&self, org: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let found = db.query_row(
"SELECT gateway FROM org WHERE space = '' AND did = ?1",
[org],
@@ -2972,7 +3114,7 @@ impl Index {
/// `hidden` those whose spaces name them taken in, where they keep a
/// record of joining.
pub fn orgs_of(&self, did: &str, hidden: bool) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(&format!(
"SELECT m.org FROM org_member m
JOIN membership j ON j.space = m.space AND j.did = m.did AND j.org = m.org
@@ -2987,7 +3129,7 @@ impl Index {
/// Where the index was last fed to from the event stream `src`.
pub fn stream_seq(&self, src: &str) -> rusqlite::Result<Option<i64>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row("SELECT seq FROM stream WHERE src = ?1", [src], |row| {
row.get(0)
})
@@ -3019,7 +3161,7 @@ impl Index {
/// Reports, newest first, from before `before` when it is given.
pub fn reports(&self, limit: usize, before: Option<i64>) -> rusqlite::Result<Vec<Value>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT id, value FROM report WHERE id < ?1 ORDER BY id DESC LIMIT ?2",
)?;
@@ -3047,7 +3189,7 @@ impl Index {
/// Whether `did` invited `runner` to its private repository `name`.
pub fn invited(&self, runner: &str, did: &str, name: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt =
db.prepare_cached("SELECT 1 FROM invite WHERE runner = ?1 AND did = ?2 AND name = ?3")?;
stmt.exists(params![runner, did, name])
@@ -3061,7 +3203,7 @@ impl Index {
after: i64,
limit: usize,
) -> rusqlite::Result<Vec<(i64, Value)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT id, did, name, at FROM invite WHERE runner = ?1 AND id > ?2 ORDER BY id LIMIT ?3",
)?;
@@ -3109,7 +3251,7 @@ impl Index {
/// `owner`'s repository `repo`'s secrets, by name, then runner.
pub fn secrets(&self, owner: &str, repo: &str) -> rusqlite::Result<Vec<Secret>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT name, runner, refs, key, sealed, at FROM secret
WHERE owner = ?1 AND repo = ?2 ORDER BY name, runner",
@@ -3119,7 +3261,7 @@ impl Index {
/// The name `did`'s repository `key` has now.
pub fn name(&self, did: &str, key: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row(
"SELECT name FROM repo WHERE did = ?1 AND key = ?2",
[did, key],
@@ -3130,7 +3272,7 @@ impl Index {
/// The record of `did`'s repository `key`.
pub fn repo_of(&self, did: &str, key: &str) -> rusqlite::Result<Option<Value>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let found = db.query_row(
"SELECT value FROM repo WHERE did = ?1 AND key = ?2",
[did, key],
@@ -3146,7 +3288,7 @@ impl Index {
pub fn repo(&self, uri: &str) -> rusqlite::Result<Option<Value>> {
let public = AtUri::parse(uri).ok().filter(|at| at.space.is_none());
let (did, name) = public.map(|at| (at.author, at.rkey)).unwrap_or_default();
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
db.query_row(
"SELECT value FROM repo WHERE uri = ?1 OR (did = ?2 AND name = ?3) LIMIT 1",
params![uri, did, name],
@@ -3174,7 +3316,7 @@ impl Index {
cursor: Option<&str>,
among: Among,
) -> rusqlite::Result<(Vec<Listed>, Option<String>)> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let pattern = format!("%{}%", query.replace(['%', '_'], ""));
let (at, uri) = after(cursor);
let mut stmt = db.prepare_cached(
@@ -3223,7 +3365,7 @@ impl Index {
/// resolves back to it, and when it was looked up, in seconds since the
/// epoch: if it ever was.
pub fn handle(&self, did: &str) -> rusqlite::Result<Option<(Option<String>, i64)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached("SELECT handle, checked FROM handle WHERE did = ?1")?;
stmt.query_row([did], |row| Ok((row.get(0)?, row.get(1)?)))
.optional()
@@ -3248,7 +3390,7 @@ impl Index {
/// The accounts owning repositories, as [`Index::repos`] lists them,
/// whose handles were never looked up.
pub fn unnamed_owners(&self) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT DISTINCT did FROM repo
WHERE did NOT IN down AND uri NOT IN down AND did NOT IN (SELECT did FROM handle)",
@@ -3265,7 +3407,7 @@ impl Index {
query: &str,
limit: usize,
) -> rusqlite::Result<Vec<(String, String, i64)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let query = query.replace(['%', '_'], "");
let mut stmt = db.prepare_cached(
"SELECT r.did, h.handle, COUNT(*) FROM repo r JOIN handle h ON h.did = r.did
@@ -3308,7 +3450,7 @@ impl Index {
Some(_) => "f.author = ?4",
None => "?4 IS NULL",
};
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
&"-- scans: forge, the blocks and hides by forge_moderation, few, and the rest
-- newest first by forge_created, or forge_author_created, to the page's limit
@@ -3368,7 +3510,7 @@ impl Index {
followed_by: Option<&str>,
quiet: bool,
) -> rusqlite::Result<Vec<Push>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let (at, uri) = after(cursor);
// From the pushes, newest first by `push_created`, or one
// account's by what it pushed and what it owns: never every
@@ -3408,7 +3550,7 @@ impl Index {
by: Option<&str>,
followed_by: Option<&str>,
) -> rusqlite::Result<Vec<Starred>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let (at, uri) = after(cursor);
// From the stars, newest first by `star_created`, or one
// account's by `star_did`: never every repository's, sorted.
@@ -3440,7 +3582,7 @@ impl Index {
since: Option<&str>,
tz: &str,
) -> rusqlite::Result<Vec<(String, String, u64)>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(&format!(
"SELECT d, kind, count(*) FROM (
SELECT zone_day(created_at, ?3) d, 'repos' kind FROM repo
@@ -3472,7 +3614,7 @@ impl Index {
since: Option<&str>,
tz: &str,
) -> rusqlite::Result<Vec<RepoPushes>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT r.name, zone_day(p.created_at, ?3), count(*), max(p.created_at),
zone_day(p.created_at, ?3) >= zone_day(coalesce(?2, strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '-1 year')), ?3)
@@ -3507,7 +3649,7 @@ impl Index {
refname: &str,
at: &str,
) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT p.after FROM push p JOIN repo r ON r.did = p.did AND r.key = p.key
WHERE r.uri = ?1 AND p.ref = ?2 AND p.created_at < ?3
@@ -3529,7 +3671,7 @@ impl Index {
at: &str,
uri: &str,
) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT COALESCE(p.after, p.before), MAX(p.created_at || ' ' || p.uri)
FROM push p JOIN repo r ON r.did = p.did AND r.key = p.key
@@ -3545,7 +3687,7 @@ impl Index {
/// Forks of `did`'s repository `key`.
pub fn forks(&self, did: &str, key: &str) -> rusqlite::Result<Vec<Listed>> {
let db = self.db.lock().unwrap();
let db = self.readers.take(&self.db);
let mut stmt = db.prepare_cached(
"SELECT did, name, description, fork_of, created_at, uri FROM repo
WHERE fork_did = ?1 AND fork_key = ?2 AND did NOT IN down AND uri NOT IN down
@@ -4093,6 +4235,13 @@ mod tests {
assert!(!kept(&mut was, &record));
index.delete("did:a", nsid::REPO, "s").unwrap();
assert!(!kept(&mut was, &record));
// An account forgotten that never wrote about it leaves it kept;
// its issue's author forgotten does not.
index.purge("did:z").unwrap();
assert!(kept(&mut was, &record));
index.purge("did:b").unwrap();
assert!(!kept(&mut was, &record));
assert!(was.items.is_empty());
// Time alone, for a label to run out.
let mut made = index.kept.lock().unwrap();
made.by.get_mut("did:a/k").unwrap().made -= KEPT_FOR;
@@ -4149,6 +4298,36 @@ mod tests {
assert_eq!(kept.order, ["d"]);
}
#[test]
fn a_read_waits_for_a_reader_to_be_free() {
let dir = tempfile::tempdir().unwrap();
let index = Index::open(&dir.path().join("i.db")).unwrap();
let mut taken: Vec<Reader> = (0..READERS)
.map(|_| index.readers.take(&index.db))
.collect();
let (said, heard) = std::sync::mpsc::channel();
std::thread::scope(|s| {
s.spawn(|| {
let db = index.readers.take(&index.db);
let one: i64 = db.query_row("SELECT 1", [], |r| r.get(0)).unwrap();
said.send(one).unwrap();
});
let waited = heard.recv_timeout(Duration::from_millis(100));
assert!(waited.is_err(), "read with every reader taken");
taken.pop();
assert_eq!(heard.recv().unwrap(), 1);
});
// A reader cannot write.
let db = index.readers.take(&index.db);
assert!(db.execute("DELETE FROM repo", []).is_err());
drop(db);
// One that cannot be opened reads through the writer.
let nowhere = Readers::new(&dir.path().join("none").join("i.db"));
let db = nowhere.take(&index.db);
let one: i64 = db.query_row("SELECT 1", [], |r| r.get(0)).unwrap();
assert_eq!(one, 1);
}
#[test]
fn stamps_past_their_number_stamp_everything() {
let stamps = Stamps::new(2);
docs/design/serve.md+10−2
View file
@@ -275,6 +275,12 @@ Details:
## The appview index
The index is written through one connection, and read through up to
eight others, in SQLite's WAL mode, each opened as reads first need it:
a page's read waits on no write, nor on another page's read, only for a
free connection once eight are reading. One that cannot be opened reads
through the writer's.
With `--index`, the server is an appview: a service that indexes records across
accounts. The SQLite index holds every public repository, and every pull,
issue, comment, state change, hide, block, queue record, check, release and
@@ -408,8 +414,10 @@ write stamps only what it is about: a record about a repository stamps
that repository, and a block stamps its owner's repositories. So the
network writing elsewhere, as it always is, leaves the rest kept. A
change whose reach is wide stamps everything at once: a takedown, a
label, a move, an account forgotten, or an account that wrote here going
inactive or coming back. Most accounts the network reports never wrote
label, a move, or an account that wrote here going inactive or coming
back. A deleted account, forgotten, stamps only the repositories it owned,
moved, wrote about or ran checks on, unless it owned a space, whose
records could be about any repository. Most accounts the network reports never wrote
here, and those stamp nothing. The record of the repository is compared
as well, since it says who may set a state or a label. And what is kept
lasts a minute at most, for a label that runs out with time.