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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

grain-serve scores what arrives with the classifiers, on Clef

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:classify/clef

Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.

crates/grain-serve/src/classify/classifiers.rsnew+471−0
View file
471 lines changed: not loaded until asked for.
crates/grain-serve/src/classify/clef.rsnew+304−0
View file
@@ -0,0 +1,304 @@
//! Clef, Cloudflare's decision model on Workers AI: asked a set of narrow
//! typed questions about a state, and the images beside it, it answers
//! each with probabilities.
//!
//! A request is `{state, model, questions, images?}`, posted with the
//! account's Workers AI token; the answer comes wrapped as Workers AI
//! wraps every model's, `{result, success, errors}`. Each answer is made a
//! score from 0 to 1 ([`Signal`]), whatever its type, so that a classifier
//! may hold a mixed set of questions to one threshold.
use serde_json::{Value, json};
use std::time::Duration;
/// Where Clef is asked, and as whom.
#[derive(Clone, Default, PartialEq)]
pub struct Config {
/// The model's URL: Workers AI's for `CLOUDFLARE_ACCOUNT_ID` and the
/// model, unless `GRAIN_CLEF_URL` names another, as tests do.
pub url: String,
/// `clef`, unless `GRAIN_CLEF_MODEL` says `clef-flash`. The thresholds
/// were tuned on `clef`, and do not carry to the other.
pub model: String,
/// `CLOUDFLARE_AI_TOKEN`: a token for Workers AI and nothing else.
pub token: String,
}
/// Its token is never written out.
impl std::fmt::Debug for Config {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Config")
.field("url", &self.url)
.field("model", &self.model)
.finish_non_exhaustive()
}
}
impl Config {
/// Clef as the environment names it, if it names an account and a
/// token. `CLOUDFLARE_API_TOKEN` is not read: on a host that also
/// manages DNS, that is the token that writes zones.
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Option<Config> {
let set = |name: &str| get(name).filter(|v| !v.is_empty());
let (account, token) = (set("CLOUDFLARE_ACCOUNT_ID")?, set("CLOUDFLARE_AI_TOKEN")?);
let model = set("GRAIN_CLEF_MODEL").unwrap_or_else(|| "clef".into());
let url = set("GRAIN_CLEF_URL").unwrap_or_else(|| {
format!(
"https://api.cloudflare.com/client/v4/accounts/{account}/ai/run/@cf/cloudflare/{model}"
)
});
Some(Config { url, model, token })
}
}
/// An answer, made a score from 0 to 1.
#[derive(Debug, Clone, PartialEq)]
pub struct Signal {
pub score: f64,
/// `noul`, `choice`, `score`, or `local` for one no model gave.
pub kind: String,
/// A choice's option, or a score's level.
pub choice: Option<String>,
pub confidence: Option<f64>,
}
impl Signal {
/// A score of `score`, from no model.
pub fn local(score: f64) -> Signal {
Signal {
score,
kind: "local".into(),
choice: None,
confidence: None,
}
}
/// As the classification keeps it.
pub fn json(&self) -> Value {
let mut out = json!({ "score": self.score, "type": self.kind });
if let Some(choice) = &self.choice {
out["choice"] = choice.as_str().into();
}
if let Some(confidence) = self.confidence {
out["confidence"] = confidence.into();
}
out
}
/// `answer`, as Clef gives one: a Noul is already a probability; a
/// Choice is the probability of the option it picked, and a Score its
/// level across the levels, each kept beside the number, which alone
/// loses which was picked.
pub fn from_answer(answer: &Value) -> Signal {
let kind = answer["type"].as_str().unwrap_or("unknown").to_string();
let confidence = answer["confidence"].as_f64();
match kind.as_str() {
"noul" => Signal {
score: answer["noul"].as_f64().unwrap_or(0.0),
kind,
choice: None,
confidence: None,
},
"choice" => {
let choice = answer["choice"].as_str().unwrap_or_default();
Signal {
score: answer["probabilities"][choice].as_f64().unwrap_or(0.0),
choice: Some(choice.into()),
kind,
confidence,
}
}
"score" => {
let levels = answer["probabilities"].as_object().map_or(0, |p| p.len());
let raw = answer["score"].as_f64().unwrap_or(0.0);
Signal {
score: if levels > 1 {
raw / (levels - 1) as f64
} else {
raw
},
choice: Some(answer["score"].to_string()),
kind,
confidence,
}
}
_ => Signal {
score: 0.0,
kind,
choice: None,
confidence: None,
},
}
}
}
/// What Clef answered: each question's signal, by its key, the model that
/// answered, and the tokens the request took.
#[derive(Debug, Clone, PartialEq)]
pub struct Answers {
pub signals: Vec<(String, Signal)>,
pub model: String,
pub input_tokens: i64,
}
/// How often, and how far apart, a request is tried.
#[derive(Debug, Clone, Copy)]
pub struct Tries {
pub attempts: u32,
/// The wait after the first failure, doubled after each one after.
pub backoff: Duration,
}
impl Default for Tries {
fn default() -> Tries {
Tries {
attempts: 4,
backoff: Duration::from_millis(500),
}
}
}
/// Statuses a later try may get past.
fn retryable(status: u16) -> bool {
matches!(status, 408 | 429 | 500 | 502 | 503 | 504)
}
/// Asks Clef `questions` of `state`, with `images`, each a `data:` URL,
/// as Clef takes them inline. A 408, 429 or 5xx, or no answer at all, is
/// tried again, after `retry-after` where the answer says, up to
/// `tries.attempts` in all; anything else Clef refuses is final.
pub async fn ask(
http: &reqwest::Client,
config: &Config,
state: &Value,
questions: &Value,
images: &[String],
tries: Tries,
) -> anyhow::Result<Answers> {
let mut body = json!({ "state": state, "model": config.model, "questions": questions });
if !images.is_empty() {
body["images"] = images.into();
}
let mut last = anyhow::anyhow!("Clef was not asked");
for attempt in 0..tries.attempts {
let wait = tries.backoff * 2u32.pow(attempt);
let sent = http
.post(&config.url)
.bearer_auth(&config.token)
.json(&body)
.send()
.await;
let res = match sent {
Ok(res) => res,
Err(e) => {
last = anyhow::anyhow!("Clef: {e}");
tokio::time::sleep(wait).await;
continue;
}
};
let status = res.status().as_u16();
if !res.status().is_success() {
let after = res.headers().get("retry-after");
let after = after.and_then(|v| v.to_str().ok()?.parse::<u64>().ok());
let text: String = res
.text()
.await
.unwrap_or_default()
.chars()
.take(200)
.collect();
last = anyhow::anyhow!("Clef {status}: {text}");
if !retryable(status) {
return Err(last);
}
let after = after.filter(|s| *s > 0).map(Duration::from_secs);
tokio::time::sleep(after.unwrap_or(wait)).await;
continue;
}
let answer: Value = res.json().await?;
if answer["success"] == false {
anyhow::bail!("Clef: {}", answer["errors"]);
}
let out = answer.get("result").unwrap_or(&answer);
let signals = out["answers"].as_object().into_iter().flatten();
let signals = signals
.map(|(key, a)| (key.clone(), Signal::from_answer(a)))
.collect();
return Ok(Answers {
signals,
model: out["model"].as_str().unwrap_or(&config.model).to_string(),
input_tokens: out["usage"]["input_tokens"].as_i64().unwrap_or(0),
});
}
Err(last)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn clef_is_named_by_the_environment() {
let vars = |pairs: &'static [(&str, &str)]| {
move |name: &str| {
let found = pairs.iter().find(|(k, _)| *k == name);
found.map(|(_, v)| v.to_string())
}
};
assert_eq!(Config::from_vars(vars(&[])), None);
assert_eq!(
Config::from_vars(vars(&[("CLOUDFLARE_ACCOUNT_ID", "acct")])),
None
);
let config = Config::from_vars(vars(&[
("CLOUDFLARE_ACCOUNT_ID", "acct"),
("CLOUDFLARE_AI_TOKEN", "secret-token"),
]))
.unwrap();
assert_eq!(
config.url,
"https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef"
);
assert_eq!(config.model, "clef");
assert!(!format!("{config:?}").contains("secret-token"));
let named = Config::from_vars(vars(&[
("CLOUDFLARE_ACCOUNT_ID", "acct"),
("CLOUDFLARE_AI_TOKEN", "t"),
("GRAIN_CLEF_MODEL", "clef-flash"),
("GRAIN_CLEF_URL", "http://127.0.0.1:1/clef"),
]))
.unwrap();
assert_eq!(
(named.url.as_str(), named.model.as_str()),
("http://127.0.0.1:1/clef", "clef-flash")
);
}
#[test]
fn every_answer_is_a_score() {
let noul = Signal::from_answer(&json!({"type": "noul", "noul": 0.8}));
assert_eq!((noul.score, noul.kind.as_str()), (0.8, "noul"));
let choice = Signal::from_answer(&json!({
"type": "choice", "choice": "b", "confidence": 0.9,
"probabilities": {"a": 0.1, "b": 0.7}
}));
assert_eq!(choice.score, 0.7);
assert_eq!(choice.choice.as_deref(), Some("b"));
let score = Signal::from_answer(&json!({
"type": "score", "score": 2, "probabilities": {"0": 0.1, "1": 0.2, "2": 0.3, "3": 0.4}
}));
assert!((score.score - 2.0 / 3.0).abs() < 1e-9);
let flat = Signal::from_answer(&json!({"type": "score", "score": 1}));
assert_eq!(flat.score, 1.0);
let odd = Signal::from_answer(&json!({"oops": true}));
assert_eq!((odd.score, odd.kind.as_str()), (0.0, "unknown"));
assert_eq!(
choice.json(),
json!({"score": 0.7, "type": "choice", "choice": "b", "confidence": 0.9})
);
assert_eq!(
Signal::local(1.0).json(),
json!({"score": 1.0, "type": "local"})
);
}
}
crates/grain-serve/src/classify/mod.rsnew+384−0
View file
@@ -0,0 +1,384 @@
//! The moderation classifiers: what is new on grain, scored as it
//! arrives, off the path of the request or the stream that brought it.
//!
//! A record a live path puts new in the index ([`Index::arrived`]) is
//! scored by the classifiers of its collection ([`classifiers::ALL`]), and
//! its author, once they have written nothing for a minute, by the
//! classifiers of accounts: an import writes hundreds of photos in a row,
//! and an account's state counts them, so the account is asked about once
//! it settles rather than once a photo. Backfill and the import do not
//! say what they put; what existed before is scored by no one here.
//!
//! Scoring a subject builds the state its classifier shows the model,
//! skips it where there is none, or where the same classifier scored the
//! same state before (its fingerprint, [`Index::fingerprint`]), and
//! otherwise asks Clef ([`clef`]) and keeps the scores
//! ([`Index::classified`]). Nothing acts on a score: one at or past its
//! threshold files a report ([`Index::report_once`]), by
//! `system:<classifier>`, into the queue a person's report lands in, and a
//! person makes every decision.
//!
//! Nothing of an account taken down, or gone from its host, is scored,
//! and nothing of one that a report or a label ever said was about the
//! safety of children ([`Index::may_classify`]): that is for a person to
//! judge, and is never handed to a model.
//!
//! Bridged accounts are the exception that acts: a gallery, story or
//! comment Bridgy Fed writes for one is labeled `spam`, which hides it, as
//! it arrives, by grain's own labeler, unless a label of `spam` was ever
//! given it, so that a moderator who took one off is not overruled. An
//! account is bridged by its handle or profile ([`classifiers::bridged`]),
//! or else by the PDS its DID document names, Bridgy Fed's: an account
//! new to the index has no handle there yet, and its profile may come
//! after its first post. That asks no model, and runs with or without
//! Clef.
//!
//! Images go to Clef inline, at most four a request, each the CDN's
//! thumbnail of its blob made at most 512 pixels on its longer side and
//! JPEG: Clef counts its context on the encoded bytes, and a larger photo
//! can overflow it.
pub mod classifiers;
pub mod clef;
use crate::cdn::{Cdn, THUMBNAIL};
use crate::index::{Index, record_uri, records};
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use classifiers::{ALL, BRIDGED_HIDDEN, Classifier, Kind, Subject};
use grain::env::Ctx;
use serde_json::json;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::{Semaphore, mpsc};
/// How many arrivals wait to be scored before more are dropped.
pub const QUEUED: usize = 10_000;
/// How many subjects are scored at once.
const AT_ONCE: usize = 4;
/// The longest side of an image shown to Clef.
pub const IMAGE_EDGE: u32 = 512;
/// How many images a request may carry.
const MOST_IMAGES: usize = 4;
/// How long things take.
#[derive(Debug, Clone, Copy)]
pub struct Timing {
/// How long an account waits after its last write to be scored.
pub settle: Duration,
/// How long a subject that failed waits to be tried again, doubled
/// each time it fails.
pub retry: Duration,
/// How many times a subject is tried before it is dropped.
pub attempts: u32,
pub clef: clef::Tries,
}
impl Default for Timing {
fn default() -> Timing {
Timing {
settle: Duration::from_secs(60),
retry: Duration::from_secs(5),
attempts: 5,
clef: clef::Tries::default(),
}
}
}
/// The classifiers, at work on an index.
pub struct Classify {
pub index: Arc<Index>,
/// What Clef, the CDN and DID documents are asked with.
pub ctx: Ctx,
/// Where images are drawn from.
pub cdn: Cdn,
/// Clef, if it is set up; without it, only bridged accounts are seen to.
pub clef: Option<clef::Config>,
/// grain's labeler, by its DID, which bridged accounts' labels are
/// given by; none gives none.
pub labeler: Option<String>,
pub timing: Timing,
}
/// What scoring a subject did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Scored {
/// Nothing: out of scope, nothing to judge, or judged as it is.
Skipped,
/// Asked, kept, and whether a report was filed.
Asked { filed: bool },
}
impl Classify {
/// Starts scoring what arrives in the index, on the runtime this is
/// called on; whether it was not already.
pub fn start(self) -> bool {
let (to, from) = mpsc::channel(QUEUED);
if !self.index.classify_to(to) {
return false;
}
tokio::spawn(Arc::new(self).run(from));
true
}
async fn run(self: Arc<Self>, mut from: mpsc::Receiver<String>) {
let settling: Arc<Mutex<HashMap<String, u64>>> = Arc::default();
let turns = Arc::new(Semaphore::new(AT_ONCE));
while let Some(uri) = from.recv().await {
let Some((did, collection, _)) = record_uri(&uri) else {
continue;
};
if BRIDGED_HIDDEN.contains(&collection) {
self.clone().bridged(uri.clone(), did.to_string());
}
if self.clef.is_none() {
continue;
}
for c in ALL.iter().filter(|c| c.covers(collection)) {
self.clone()
.queue(c, uri.clone(), did.to_string(), turns.clone());
}
self.clone()
.settle(did.to_string(), settling.clone(), turns.clone());
}
}
/// Scores `did` with the account classifiers once it has written
/// nothing for [`Timing::settle`].
fn settle(
self: Arc<Self>,
did: String,
settling: Arc<Mutex<HashMap<String, u64>>>,
turns: Arc<Semaphore>,
) {
let mine = {
let mut settling = settling.lock().unwrap();
let n = settling.entry(did.clone()).or_default();
*n += 1;
*n
};
tokio::spawn(async move {
tokio::time::sleep(self.timing.settle).await;
{
let mut settling = settling.lock().unwrap();
if settling.get(&did) != Some(&mine) {
return;
}
settling.remove(&did);
}
for c in ALL.iter().filter(|c| c.kind == Kind::Account) {
self.clone()
.queue(c, did.clone(), did.clone(), turns.clone());
}
});
}
/// Scores `subject`, of the account `did`, with `c`, in turn, trying
/// again after a failure until [`Timing::attempts`].
fn queue(
self: Arc<Self>,
c: &'static Classifier,
subject: String,
did: String,
turns: Arc<Semaphore>,
) {
tokio::spawn(async move {
let scored = tried(self.timing, || async {
let _turn = turns.acquire().await.expect("never closed");
self.score(c, &subject, &did).await.map(drop)
});
if let Err(e) = scored.await {
tracing::warn!(classifier = c.name, %subject, error = %e, "classify: dropped");
}
});
}
/// Labels the post at `uri` `spam` if `did` was bridged in, apart from
/// the loop that takes arrivals, as a DID document may be slow to
/// read, trying again after a failure until [`Timing::attempts`].
fn bridged(self: Arc<Self>, uri: String, did: String) {
if self.labeler.is_none() {
return;
}
tokio::spawn(async move {
let labeled = tried(self.timing, || self.label_bridged(&uri, &did));
if let Err(e) = labeled.await {
tracing::warn!(%uri, error = %e, "classify: labeling a bridged post");
}
});
}
async fn label_bridged(&self, uri: &str, did: &str) -> anyhow::Result<()> {
let labeler = self.labeler.as_deref().expect("a labeler");
let bridged = match self.index.read(|db| classifiers::bridged(db, did))? {
true => true,
false => {
let doc = grain::identity::kept_document(&self.ctx, did).await?;
classifiers::bridged_host(&doc)
}
};
if bridged {
let now = grain::keys::now_iso();
self.index.label_once(labeler, uri, "spam", &now)?;
}
Ok(())
}
/// Scores `subject`, of the account `did`, with `c`, as it stands now.
pub async fn score(&self, c: &Classifier, subject: &str, did: &str) -> anyhow::Result<Scored> {
let Some(clef) = &self.clef else {
return Ok(Scored::Skipped);
};
if !self.index.may_classify(did)? {
return Ok(Scored::Skipped);
}
if let Some((_, collection, _)) = record_uri(subject) {
let table = records::table_of(collection).expect("a collection classified");
if !self.index.holds(table, subject)? {
return Ok(Scored::Skipped);
}
}
let handle = self.index.handle(did)?.flatten();
let s = Subject {
uri: subject.to_string(),
did: did.to_string(),
handle,
};
let Some(state) = self.index.read(|db| (c.state)(db, &s))? else {
return Ok(Scored::Skipped);
};
let fingerprint = Sha256::new()
.chain_update(c.version())
.chain_update(state.to_string())
.finalize();
let fingerprint: String = fingerprint[..16]
.iter()
.map(|b| format!("{b:02x}"))
.collect();
if self.index.fingerprint(c.name, subject)?.as_deref() == Some(&fingerprint) {
return Ok(Scored::Skipped);
}
let mut images = vec![];
for (did, cid) in (c.images)(&state).into_iter().take(MOST_IMAGES) {
let blob = json!({ "ref": { "$link": cid } });
let url = self
.cdn
.url(&did, &blob, THUMBNAIL)
.expect("a blob that names a CID");
images.push(self.image(&url).await?);
}
let answers = clef::ask(
&self.ctx.http,
clef,
&state,
&(c.questions)(),
&images,
self.timing.clef,
)
.await?;
let now = grain::keys::now_iso();
self.index
.classified(&crate::index::classify::Classification {
subject: subject.to_string(),
subject_did: did.to_string(),
classifier: c.name.to_string(),
signals: answers.signals.clone(),
state,
fingerprint,
model: answers.model,
scanned_at: now.clone(),
})?;
let mut crossing: Option<(&str, f64)> = None;
for (signal, s) in &answers.signals {
let past = c.threshold(signal).is_some_and(|t| s.score >= t);
if past && crossing.is_none_or(|(_, top)| s.score > top) {
crossing = Some((signal, s.score));
}
}
let Some((signal, score)) = crossing else {
return Ok(Scored::Asked { filed: false });
};
let reason = format!("{}: {signal} {score:.2}", c.name);
let by = format!("system:{}", c.name);
let label = (c.label)(signal);
let filed = self
.index
.report_once((subject, did), label, &reason, &by, &now)?;
if filed {
tracing::info!(classifier = c.name, %subject, label, "classify: filed a report");
}
Ok(Scored::Asked { filed })
}
/// The image at `url`, at most [`IMAGE_EDGE`] on its longer side, as
/// a JPEG `data:` URL.
async fn image(&self, url: &str) -> anyhow::Result<String> {
let res = self.ctx.http.get(url).send().await?.error_for_status()?;
let bytes = res.bytes().await?;
let jpeg = tokio::task::spawn_blocking(move || shrink(&bytes)).await??;
Ok(format!("data:image/jpeg;base64,{}", STANDARD.encode(jpeg)))
}
}
/// Runs `f` until it succeeds, [`Timing::attempts`] times at most,
/// waiting [`Timing::retry`] after the first failure, doubled after each
/// one after; the last failure, if none did.
async fn tried<F, Fut>(timing: Timing, mut f: F) -> anyhow::Result<()>
where
F: FnMut() -> Fut,
Fut: Future<Output = anyhow::Result<()>>,
{
let mut attempt = 1;
loop {
let Err(e) = f().await else {
return Ok(());
};
if attempt >= timing.attempts {
return Err(e);
}
tokio::time::sleep(timing.retry * 2u32.pow(attempt - 1)).await;
attempt += 1;
}
}
/// `bytes`, a PNG, JPEG or WebP, as a JPEG no more than [`IMAGE_EDGE`] on
/// its longer side.
pub fn shrink(bytes: &[u8]) -> anyhow::Result<Vec<u8>> {
let image = image::load_from_memory(bytes)?;
let image = match image.width().max(image.height()) > IMAGE_EDGE {
true => image.resize(
IMAGE_EDGE,
IMAGE_EDGE,
image::imageops::FilterType::Triangle,
),
false => image,
};
let mut out = std::io::Cursor::new(Vec::new());
let encoder = image::codecs::jpeg::JpegEncoder::new_with_quality(&mut out, 80);
image.to_rgb8().write_with_encoder(encoder)?;
Ok(out.into_inner())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn images_are_shrunk_to_512_as_jpeg() {
let png = |w, h| {
let mut out = std::io::Cursor::new(Vec::new());
let image = image::RgbImage::from_pixel(w, h, image::Rgb([200, 10, 10]));
image.write_to(&mut out, image::ImageFormat::Png).unwrap();
out.into_inner()
};
let big = image::load_from_memory(&shrink(&png(2000, 1000)).unwrap()).unwrap();
assert_eq!((big.width(), big.height()), (512, 256));
let small = shrink(&png(100, 300)).unwrap();
assert!(small.starts_with(&[0xFF, 0xD8, 0xFF]));
let small = image::load_from_memory(&small).unwrap();
assert_eq!((small.width(), small.height()), (100, 300));
assert!(shrink(b"<html>").is_err());
}
}
crates/grain-serve/src/index/classify.rsnew+184−0
View file
@@ -0,0 +1,184 @@
//! What the classifiers ([`crate::classify`]) read and keep in the index.
//!
//! The paths that put what accounts write as they write it (the stream,
//! `indexRecords` and the web's writes) say which records were new to
//! the index ([`Index::arrived`]), for the classifiers to score. Backfill
//! and the import do not: what they put is not new.
//!
//! `classification` keeps each subject's last scores, by classifier, and
//! the fingerprint of what was asked, so that what has not changed is not
//! asked again. It is made from the records, by asking again.
use super::Index;
use crate::classify::clef::Signal;
use crate::index::schema::shown;
use rusqlite::{OptionalExtension, params};
use serde_json::Value;
use tokio::sync::mpsc::{Sender, error::TrySendError};
/// A subject's scores, as `classification` keeps them.
#[derive(Debug, Clone, PartialEq)]
pub struct Classification {
/// An account's bare DID, or a record's AT-URI.
pub subject: String,
pub subject_did: String,
pub classifier: String,
pub signals: Vec<(String, Signal)>,
pub state: Value,
pub fingerprint: String,
pub model: String,
pub scanned_at: String,
}
impl Index {
/// Has the records [`Index::arrived`] names sent to `to` as well, from
/// then on; whether none was named before.
pub fn classify_to(&self, to: Sender<String>) -> bool {
self.classifying.set(to).is_ok()
}
/// Tells the classifiers of the record at `uri`, which a live path put
/// new in the index ([`Index::arrived`]). Where they are behind by
/// [`crate::classify::QUEUED`] records, it is not scored.
pub(super) fn to_classify(&self, uri: &str) {
let Some(to) = self.classifying.get() else {
return;
};
match to.try_send(uri.to_string()) {
Ok(()) => {}
Err(TrySendError::Full(_)) => tracing::warn!(%uri, "classify: behind, not scored"),
Err(TrySendError::Closed(_)) => tracing::warn!(%uri, "classify: stopped, not scored"),
}
}
/// The fingerprint `classifier` last scored `subject` at, if it did.
pub fn fingerprint(&self, classifier: &str, subject: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let sql = "SELECT fingerprint FROM classification WHERE classifier = ?1 AND subject = ?2";
db.query_row(sql, [classifier, subject], |r| r.get(0))
.optional()
}
/// Keeps `c`, in place of what its classifier scored its subject
/// before.
pub fn classified(&self, c: &Classification) -> rusqlite::Result<()> {
let signals: serde_json::Map<String, Value> = c
.signals
.iter()
.map(|(k, s)| (k.clone(), s.json()))
.collect();
let (top, score) = crate::classify::classifiers::top(&c.signals);
let db = self.db.lock().unwrap();
db.execute(
"INSERT INTO classification (subject, classifier, subject_did, signals, state,
top_signal, top_score, fingerprint, model, scanned_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
ON CONFLICT (subject, classifier) DO UPDATE SET subject_did = excluded.subject_did,
signals = excluded.signals, state = excluded.state,
top_signal = excluded.top_signal, top_score = excluded.top_score,
fingerprint = excluded.fingerprint, model = excluded.model,
scanned_at = excluded.scanned_at",
params![
c.subject,
c.classifier,
c.subject_did,
Value::Object(signals).to_string(),
c.state.to_string(),
top,
score,
c.fingerprint,
c.model,
c.scanned_at
],
)?;
Ok(())
}
/// The scores `classifier` gave `subject`, if it scored it.
pub fn classification(
&self,
classifier: &str,
subject: &str,
) -> rusqlite::Result<Option<(Value, Option<String>, f64)>> {
let db = self.db.lock().unwrap();
let sql = "SELECT signals, top_signal, top_score FROM classification
WHERE classifier = ?1 AND subject = ?2";
db.query_row(sql, [classifier, subject], |r| {
let signals: String = r.get(0)?;
let signals = serde_json::from_str(&signals).unwrap_or_default();
Ok((signals, r.get(1)?, r.get(2)?))
})
.optional()
}
/// Files a report on `subject`, of the account `did`, asking for
/// `label`, by `by` at `now`, unless one asking for it is open; whether
/// it filed one. A second report would only split the review.
pub fn report_once(
&self,
(subject, did): (&str, &str),
label: &str,
reason: &str,
by: &str,
now: &str,
) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let filed = db.execute(
"INSERT INTO report (subject_uri, subject_did, label, reason, reported_by, created_at)
SELECT ?1, ?2, ?3, ?4, ?5, ?6 WHERE NOT EXISTS (SELECT 1 FROM report
WHERE subject_uri = ?1 AND label = ?3 AND status = 'open')",
params![subject, did, label, reason, by, now],
)?;
Ok(filed > 0)
}
/// Gives out `src`'s `val` on `uri` at `now`, unless any label of
/// `val` was ever given it, or negated: a moderator who took one off
/// is not overruled. Whether it gave one.
pub fn label_once(&self, src: &str, uri: &str, val: &str, now: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let given = db.execute(
"INSERT INTO label (src, uri, val, neg, cts)
SELECT ?1, ?2, ?3, 0, ?4 WHERE NOT EXISTS (SELECT 1 FROM label
WHERE uri = ?2 AND val = ?3)",
params![src, uri, val, now],
)?;
Ok(given > 0)
}
/// Whether what the account `did` writes may be shown to a model: it
/// is shown here, and nothing of it was ever reported or labeled as
/// being about the safety of children (`csae`), by grain, a reporter
/// or its author, negated since or not. What a report or label of that
/// is about is for a person to judge, and is never handed to a model.
pub fn may_classify(&self, did: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let ours = "(uri = ?1 OR (uri > 'at://' || ?1 || '/' AND uri < 'at://' || ?1 || '0'))";
let sql = format!(
"SELECT {shown}
AND NOT EXISTS (SELECT 1 FROM label WHERE val = 'csae' AND {ours})
AND NOT EXISTS (SELECT 1 FROM self_label WHERE val = 'csae' AND {ours})
AND NOT EXISTS (SELECT 1 FROM report WHERE label = 'csae' AND (subject_did = ?1
OR subject_uri = ?1 OR (subject_uri > 'at://' || ?1 || '/'
AND subject_uri < 'at://' || ?1 || '0')))",
shown = shown("?1"),
);
db.query_row(&sql, [did], |r| r.get(0))
}
/// The handle of `did`, if the index tracks it, as it keeps it.
pub fn handle(&self, did: &str) -> rusqlite::Result<Option<Option<String>>> {
let db = self.db.lock().unwrap();
db.query_row("SELECT handle FROM repo WHERE did = ?1", [did], |r| {
r.get(0)
})
.optional()
}
/// Whether the index holds the record at `uri`, of `table`.
pub fn holds(&self, table: &str, uri: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let sql = format!("SELECT 1 FROM {table} WHERE uri = ?1");
db.prepare(&sql)?.exists([uri])
}
}
crates/grain-serve/src/index/mod.rs+5−0
View file
@@ -14,6 +14,7 @@
//! leaves alone what the index has from a later one ([`Applied::Stale`]).
pub mod backfill;
pub mod classify;
#[cfg(test)]
pub mod plan;
pub mod preferences;
@@ -40,6 +41,9 @@ pub struct Index {
/// Where the records live paths newly put are told, for push
/// ([`Index::arrived`]).
arrivals: std::sync::OnceLock<tokio::sync::mpsc::Sender<String>>,
/// Where the same records are told for the classifiers
/// ([`Index::classify_to`]).
classifying: std::sync::OnceLock<tokio::sync::mpsc::Sender<String>>,
}
/// What putting a record did.
@@ -204,6 +208,7 @@ impl Index {
lexicons: Lexicons::grain(),
tracked: Mutex::new(tracked),
arrivals: std::sync::OnceLock::new(),
classifying: std::sync::OnceLock::new(),
})
}
crates/grain-serve/src/index/push.rs+4−2
View file
@@ -32,9 +32,11 @@ impl Index {
}
/// Says that a live path put the record at `uri`, which the index did
/// not hold, for it to be pushed. Where the sender is behind by
/// [`crate::push::send::QUEUED`] records, it is not.
/// not hold, for it to be pushed, and scored by the classifiers
/// ([`Index::to_classify`]). Where the sender is behind by
/// [`crate::push::send::QUEUED`] records, it is not pushed.
pub fn arrived(&self, uri: &str) {
self.to_classify(uri);
let Some(to) = self.arrivals.get() else {
return;
};
crates/grain-serve/src/index/schema.rs+23−1
View file
@@ -84,6 +84,10 @@
//! login was kept with none before it, in microseconds since the epoch:
//! a cookie signed before then names no one.
//!
//! `classification` is neither: the classifiers' last scores of each
//! subject ([`crate::index::classify`]), which scoring again would make,
//! at the cost of asking Clef again.
//!
//! A migration may drop and remake a record table, which backfill refills;
//! these it only alters in place.
//!
@@ -103,7 +107,7 @@
/// Each step of the schema, run once, in order, by `PRAGMA user_version`.
/// A step is only ever appended: one that has run is never changed.
pub const MIGRATIONS: &[&str] = &[V1, V2, V3, V4, V5, V6, V7, V8, V9];
pub const MIGRATIONS: &[&str] = &[V1, V2, V3, V4, V5, V6, V7, V8, V9, V10];
const V1: &str = "
CREATE TABLE gallery (
@@ -482,3 +486,21 @@ const V9: &str = "
CREATE TABLE push_record (uri TEXT PRIMARY KEY, at TEXT NOT NULL) WITHOUT ROWID;
CREATE INDEX push_record_at ON push_record (at);
";
/// What the classifiers scored ([`crate::index::classify`]).
const V10: &str = "
CREATE TABLE classification (
subject TEXT NOT NULL,
classifier TEXT NOT NULL,
subject_did TEXT NOT NULL,
signals TEXT NOT NULL,
state TEXT,
top_signal TEXT,
top_score REAL NOT NULL DEFAULT 0,
fingerprint TEXT NOT NULL,
model TEXT,
scanned_at TEXT NOT NULL,
PRIMARY KEY (subject, classifier)
) WITHOUT ROWID;
CREATE INDEX classification_did ON classification (subject_did);
";
crates/grain-serve/src/lib.rs+28−1
View file
@@ -3,10 +3,13 @@
//! the index alone ([`read`]), what an account has seen of its
//! notifications ([`notification`]) and the devices they are pushed to
//! ([`push`]), and the web's writes ([`write`]), draws the cards links
//! unfurl into ([`og`]), and logs the web in ([`login`]).
//! unfurl into ([`og`]), and logs the web in ([`login`]). Off the path of
//! any request, its classifiers score what arrives for moderators to
//! review ([`classify`]).
pub mod backfill;
pub mod cdn;
pub mod classify;
pub mod index;
pub mod jetstream;
pub mod login;
@@ -65,6 +68,11 @@ pub struct Settings {
/// the services notifications are pushed by. With neither, as tests
/// and dev run, none is pushed.
pub push: push::Config,
/// `CLOUDFLARE_ACCOUNT_ID` and `CLOUDFLARE_AI_TOKEN`, and the rest
/// ([`classify::clef::Config::from_vars`]): Clef, for the classifiers
/// to ask. None asks no model, and only bridged accounts' posts are
/// labeled.
pub clef: Option<classify::clef::Config>,
}
impl Settings {
@@ -88,6 +96,7 @@ impl Settings {
sign_up_pds: set("GRAIN_SIGN_UP_PDS"),
carry_cookies_until: set("GRAIN_CARRY_COOKIES_UNTIL"),
push: push::Config::from_vars(&get),
clef: classify::clef::Config::from_vars(&get),
}
}
@@ -138,6 +147,24 @@ impl Settings {
Ok(true)
}
/// Starts the classifiers on `app`'s index, on the runtime this is
/// called on, asking the Clef these settings name; whether there is an
/// index for them to score.
pub fn classify(&self, app: &App) -> bool {
let Some(index) = &app.index else {
return false;
};
let classify = classify::Classify {
index: index.clone(),
ctx: app.ctx.clone(),
cdn: app.cdn.clone(),
clef: self.clef.clone(),
labeler: app.did.clone(),
timing: classify::Timing::default(),
};
classify.start()
}
/// The server `ctx` and these settings make: its index opened, and
/// its DID named.
pub fn app(&self, ctx: Ctx) -> anyhow::Result<App> {
crates/grain-serve/src/main.rs+1−0
View file
@@ -20,6 +20,7 @@ async fn main() -> anyhow::Result<()> {
tokio::spawn(backfill.run());
}
settings.push(&app)?;
settings.classify(&app);
settings.follow(&app)?;
let addr = std::env::var("GRAIN_LISTEN").unwrap_or_else(|_| "127.0.0.1:2600".into());
let listener = tokio::net::TcpListener::bind(&addr)
crates/grain-serve/tests/classify.rsnew+788−0
View file
788 lines changed: not loaded until asked for.
crates/grain-serve/Cargo.toml+1−0
View file
@@ -14,6 +14,7 @@ axum.workspace = true
base64.workspace = true
bytes.workspace = true
futures.workspace = true
image.workspace = true
p256.workspace = true
png.workspace = true
reqwest.workspace = true
Cargo.lock+33−0
View file
@@ -815,6 +815,7 @@ dependencies = [
"bytes",
"futures",
"grain",
"image",
"p256",
"png",
"reqwest",
@@ -1142,6 +1143,22 @@ dependencies = [
"icu_properties",
]
[[package]]
name = "image"
version = "0.25.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85ab80394333c02fe689eaf900ab500fbd0c2213da414687ebf995a65d5a6104"
dependencies = [
"bytemuck",
"byteorder-lite",
"image-webp",
"moxcms",
"num-traits",
"png",
"zune-core",
"zune-jpeg",
]
[[package]]
name = "image-webp"
version = "0.2.4"
@@ -1341,6 +1358,16 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "moxcms"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bb85c154ba489f01b25c0d36ae69a87e4a1c73a72631fc6c0eb6dde34a73e44b"
dependencies = [
"num-traits",
"pxfm",
]
[[package]]
name = "num-traits"
version = "0.2.19"
@@ -1487,6 +1514,12 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "pxfm"
version = "0.1.30"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d55d956fa96f5ec02be2e13af0e20391a5aa83d6a074e3ad368959d0fab299ea"
[[package]]
name = "quick-error"
version = "2.0.1"
Cargo.toml+1−0
View file