chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Profiles answer at once: badges read pulls by index, and down is a table
crates/knotless-serve/src/index.rs+94−11
| @@ -487,11 +487,21 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> { | ||
sealed BLOB NOT NULL, at TEXT NOT NULL, | ||
PRIMARY KEY (owner, repo, runner, name)); | ||
CREATE TABLE IF NOT EXISTS inactive (did TEXT PRIMARY KEY, status TEXT NOT NULL); | ||
CREATE TEMP VIEW IF NOT EXISTS down AS | ||
SELECT subject FROM takedown | ||
UNION SELECT did FROM inactive | ||
UNION SELECT uri FROM label | ||
WHERE exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now');" | ||
CREATE INDEX IF NOT EXISTS label_uri ON label (uri); | ||
DROP VIEW IF EXISTS temp.down; | ||
CREATE TABLE IF NOT EXISTS down (subject TEXT PRIMARY KEY); | ||
{triggers}", | ||
triggers = down_triggers(), | ||
))?; | ||
// What is down, from what makes it so, as it stands now: the | ||
// triggers keep it so from here, but for labels as they lapse. | ||
db.execute_batch(&format!( | ||
"BEGIN; | ||
DELETE FROM down; | ||
INSERT OR IGNORE INTO down SELECT subject FROM takedown | ||
UNION ALL SELECT did FROM inactive | ||
UNION ALL SELECT uri FROM label WHERE {LIVE}; | ||
COMMIT;" | ||
))?; | ||
// Indexes of what records say, apart: a table of another shape, | ||
// with no `value`, is kept without them, as every use of it fails. | ||
| @@ -567,6 +577,44 @@ const CACHE: &str = "DROP TABLE IF EXISTS repo; DROP TABLE IF EXISTS forge; | ||
DROP TABLE IF EXISTS org; DROP TABLE IF EXISTS org_member; DROP TABLE IF EXISTS membership; | ||
DROP TABLE IF EXISTS collaborator;"; | ||
| ||
/// A label that has not lapsed. | ||
const LIVE: &str = "(exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"; | ||
| ||
/// Whether nothing keeps `down.subject` down any more: no takedown, no | ||
/// host saying its account is inactive, and no label that has not lapsed. | ||
fn up() -> String { | ||
format!( | ||
"NOT EXISTS (SELECT 1 FROM takedown t WHERE t.subject = down.subject) | ||
AND NOT EXISTS (SELECT 1 FROM inactive i WHERE i.did = down.subject) | ||
AND NOT EXISTS (SELECT 1 FROM label l WHERE l.uri = down.subject AND {})", | ||
LIVE.replace("exp", "l.exp") | ||
) | ||
} | ||
| ||
/// What keeps `down`, a table of every subject left out of reads, as | ||
/// takedowns, accounts' hosts and labelers say: a table, not a view of | ||
/// them, so that `NOT IN down` looks each up by its key rather than | ||
/// building the whole list for every statement, as the network's | ||
/// inactive accounts are many. A subject goes when nothing keeps it down. | ||
fn down_triggers() -> String { | ||
let unless = up(); | ||
let live = LIVE.replace("exp", "new.exp"); | ||
format!( | ||
"CREATE TRIGGER IF NOT EXISTS takedown_down AFTER INSERT ON takedown | ||
BEGIN INSERT OR IGNORE INTO down VALUES (new.subject); END; | ||
CREATE TRIGGER IF NOT EXISTS inactive_down AFTER INSERT ON inactive | ||
BEGIN INSERT OR IGNORE INTO down VALUES (new.did); END; | ||
CREATE TRIGGER IF NOT EXISTS label_down AFTER INSERT ON label WHEN {live} | ||
BEGIN INSERT OR IGNORE INTO down VALUES (new.uri); END; | ||
CREATE TRIGGER IF NOT EXISTS takedown_up AFTER DELETE ON takedown | ||
BEGIN DELETE FROM down WHERE subject = old.subject AND {unless}; END; | ||
CREATE TRIGGER IF NOT EXISTS inactive_up AFTER DELETE ON inactive | ||
BEGIN DELETE FROM down WHERE subject = old.did AND {unless}; END; | ||
CREATE TRIGGER IF NOT EXISTS label_up AFTER DELETE ON label | ||
BEGIN DELETE FROM down WHERE subject = old.uri AND {unless}; END;" | ||
) | ||
} | ||
| ||
/// Who repositories' records name among their collaborators, a row each, | ||
/// from their rows in `repo`: the one at `?1`, or every one when it is | ||
/// null. | ||
| @@ -1240,18 +1288,27 @@ impl Index { | ||
/// taken down, is left out. | ||
fn pulls_where(&self, which: &str, did: &str) -> rusqlite::Result<Vec<Pulls>> { | ||
let db = self.db.lock().unwrap(); | ||
// 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. | ||
let mut stmt = db.prepare_cached(&format!( | ||
"SELECT f.uri, f.author, f.value, r.did, r.value, r.uri | ||
"SELECT p.uri, p.author, p.value, r.did, r.value, r.uri | ||
FROM forge p | ||
JOIN repo r ON r.did = p.repo_did AND r.key = p.repo_key | ||
JOIN forge f ON f.repo_did = p.repo_did AND f.repo_key = p.repo_key | ||
AND (f.uri = p.uri OR ((instr(f.uri, '/dev.knotless.state/') | ||
OR instr(f.uri, '/dev.knotless.queue/')) | ||
AND f.value ->> '$.subject' = p.uri)) | ||
WHERE {which} | ||
AND instr(p.uri, '/dev.knotless.pull/') | ||
AND p.uri NOT IN down AND r.uri NOT IN down | ||
ORDER BY r.uri" | ||
UNION ALL | ||
SELECT f.uri, f.author, f.value, r.did, r.value, r.uri | ||
FROM forge p | ||
CROSS JOIN repo r ON r.did = p.repo_did AND r.key = p.repo_key | ||
CROSS JOIN forge f ON json_extract(f.value, '$.subject') = +p.uri | ||
AND +f.repo_did = p.repo_did AND +f.repo_key = p.repo_key | ||
WHERE {which} | ||
AND instr(p.uri, '/dev.knotless.pull/') | ||
AND p.uri NOT IN down AND r.uri NOT IN down | ||
AND (instr(f.uri, '/dev.knotless.state/') OR instr(f.uri, '/dev.knotless.queue/')) | ||
ORDER BY 6" | ||
))?; | ||
let rows = stmt.query_map([did], elsewhere)?; | ||
let mut by_repo: Vec<(String, String, String, Vec<Entry>)> = Vec::new(); | ||
| @@ -2193,6 +2250,24 @@ impl Index { | ||
Ok(()) | ||
} | ||
| ||
/// Puts back what was down only for labels that have since lapsed, | ||
/// as `down` keeps them until this is asked. | ||
pub fn lapse(&self) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
let lapsed = db.execute( | ||
&format!( | ||
"DELETE FROM down WHERE subject IN (SELECT uri FROM label WHERE NOT {LIVE}) | ||
AND {}", | ||
up() | ||
), | ||
[], | ||
)?; | ||
if lapsed > 0 { | ||
self.stamps.mark_all(); | ||
} | ||
Ok(()) | ||
} | ||
| ||
/// 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(); | ||
| @@ -4544,6 +4619,14 @@ mod tests { | ||
.unwrap(); | ||
assert_eq!(forge().len(), 2); | ||
assert_eq!(index.labeler_seq("did:web:l").unwrap(), Some(7)); | ||
// One that lapses hides until a sweep puts it back. | ||
for (seq, exp) in [(8, "2999-01-01T00:00:00Z"), (9, "2000-01-01T00:00:00Z")] { | ||
let hide = label("did:web:l", "did:c", "!hide", json!({"exp": exp})); | ||
index.label("did:web:l", seq, &hide).unwrap(); | ||
assert_eq!(forge(), std::slice::from_ref(&first)); | ||
} | ||
index.lapse().unwrap(); | ||
assert_eq!(forge().len(), 2); | ||
| ||
// Reports, newest first, a page at a time. | ||
for n in 0..3 { | ||
crates/knotless-serve/src/lib.rs+2−0
| @@ -734,9 +734,11 @@ const SUMMARY: Duration = Duration::from_secs(60); | ||
/// Backfills from `relay` every `every`, until the process ends. A record | ||
/// whose event a feed never delivered, one Jetstream dropped or a commit | ||
/// the relay resynced over, is otherwise missing from the index for good. | ||
/// And puts back, each time, what labels that lapsed took down. | ||
pub async fn sweep(ctx: Ctx, index: Arc<index::Index>, relay: String, every: Duration) { | ||
loop { | ||
tokio::time::sleep(every).await; | ||
let _ = index.lapse(); | ||
telemetry::sweep(&ctx, &index, &relay).await; | ||
} | ||
} | ||