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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

server: Jetstream feeds the index, resuming where it was read to

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:index/jetstream

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

crates/grain/src/jetstream.rsnew+428−0
View file
428 lines changed: not loaded until asked for.
crates/grain/src/lib.rs+1−0
View file
@@ -10,6 +10,7 @@ pub mod crypto;
pub mod drisl;
pub mod env;
pub mod identity;
pub mod jetstream;
pub mod keys;
pub mod lexicon;
pub mod oauth;
crates/grain/Cargo.toml+1−0
View file
@@ -21,6 +21,7 @@ serde.workspace = true
serde_json.workspace = true
sha2.workspace = true
tokio.workspace = true
tokio-tungstenite.workspace = true
tracing.workspace = true
unicode-segmentation.workspace = true
zstd.workspace = true
crates/grain-serve/src/index/mod.rs+8−0
View file
@@ -439,6 +439,14 @@ impl Index {
Ok(())
}
/// Drops where the stream `source` was read to, so that it is next
/// read from its live tip.
pub fn forget_cursor(&self, source: &str) -> rusqlite::Result<()> {
let db = self.db.lock().unwrap();
db.execute("DELETE FROM stream_cursor WHERE source = ?1", [source])?;
Ok(())
}
/// Gives out `label`; its sequence number.
pub fn label(&self, label: &Label) -> rusqlite::Result<i64> {
let db = self.db.lock().unwrap();
crates/grain-serve/src/index/tests.rs+3−0
View file
@@ -374,6 +374,9 @@ fn accounts() {
index.set_cursor(source, 41).unwrap();
index.set_cursor(source, 42).unwrap();
assert_eq!(index.cursor(source).unwrap(), Some(42));
index.forget_cursor(source).unwrap();
assert_eq!(index.cursor(source).unwrap(), None);
index.set_cursor(source, 42).unwrap();
// An account's label is on its bare DID; a record's on its URI.
let label = |uri: &str, neg: bool| Label {
src: "did:plc:grain".into(),
crates/grain-serve/src/jetstream.rsnew+192−0
View file
@@ -0,0 +1,192 @@
//! The index fed from Jetstream: commits to the collections it keeps, by
//! the accounts it tracks, and what hosts say of those accounts.
//!
//! As hatk's indexer: an account is tracked once it writes a grain profile
//! or a group declaration (the signal collections), and the stream's other
//! records are kept only of tracked accounts, so the Bluesky posts of the
//! whole network pass by. A record that does not fit its lexicon is
//! skipped, not fatal. So is an event the index fails to apply: it is
//! logged and the stream goes on past it, and once the cursor is saved
//! that event is not heard again.
//!
//! A commit is written at its revision ([`Index::put_rev`]), so that a
//! stream write and a backfill's read of the same repo leave the later of
//! the two. An account newly tracked, or back after its host said it was
//! gone, is queued for backfill ([`Index::resync`]) to be read whole.
//!
//! Where it was read to is kept in the index, under `jetstream:<url>` as
//! hatk's `_cursor` kept it, and a restart resumes there. A cursor
//! Jetstream no longer keeps is given up for the live tip
//! ([`grain::jetstream::PROBE_EVERY`]).
use crate::backfill::SIGNALS;
use crate::index::{Index, PutError, records};
use grain::jetstream::{Event, Follower, Timing, What};
use std::sync::Arc;
use std::time::{Duration, Instant};
/// The Jetstream grain follows unless told another.
pub const DEFAULT: &str = "wss://jetstream.us-west.bsky.network";
/// How often where the stream was read to is saved, at most.
pub const SAVE: Duration = Duration::from_secs(5);
/// The cursor's key in the index for the Jetstream at `url`: its own, as
/// one instance's sequence numbers mean nothing on another.
pub fn source(url: &str) -> String {
format!("jetstream:{url}")
}
/// Follows the Jetstream at `url` into `index` until the task is dropped,
/// saving where it was read to at most every `save`.
pub async fn follow(index: Arc<Index>, url: String, timing: Timing, save: Duration) {
let collections = records::COLLECTIONS.iter().map(|(nsid, _)| *nsid).collect();
let feed = Feed {
source: source(&url),
index,
seq: None,
saved: None,
save,
};
grain::jetstream::follow(url, collections, timing, feed).await
}
struct Feed {
index: Arc<Index>,
source: String,
/// The last sequence number heard.
seq: Option<i64>,
saved: Option<Instant>,
save: Duration,
}
impl Feed {
fn save(&mut self) -> rusqlite::Result<()> {
if let Some(seq) = self.seq {
self.index.set_cursor(&self.source, seq)?;
self.saved = Some(Instant::now());
}
Ok(())
}
}
/// Logs an index that failed: the stream goes on.
fn logged(done: rusqlite::Result<()>) {
if let Err(e) = done {
tracing::error!(error = %e, "jetstream: the index failed");
}
}
impl Follower for Feed {
fn start(&mut self) -> Option<i64> {
self.seq = self.index.cursor(&self.source).ok().flatten();
tracing::info!(cursor = ?self.seq, source = %self.source, "jetstream: following");
self.seq
}
fn connected(&mut self, up: bool) {
tracing::info!(up, source = %self.source, "jetstream: connecting");
}
fn event(&mut self, event: Event) {
self.seq = event.seq.or(self.seq);
logged(apply(&self.index, event));
if self.saved.is_none_or(|at| at.elapsed() >= self.save) {
logged(self.save());
}
}
fn ended(&mut self) {
logged(self.save());
}
fn cursor_lost(&mut self, cursor: i64) {
self.seq = None;
logged(self.index.forget_cursor(&self.source));
tracing::warn!(cursor, source = %self.source,
"jetstream: the cursor is past what Jetstream keeps; following from the live tip, \
and what came between is not heard");
}
}
/// What `event` says, applied to `index`.
pub fn apply(index: &Index, event: Event) -> rusqlite::Result<()> {
match event.what {
What::Commit { did, op } => {
if records::table_of(&op.collection).is_none() {
return Ok(());
}
// An account newly tracked is read whole by backfill.
if SIGNALS.contains(&op.collection.as_str()) && index.track(&did)? {
index.resync(&did)?;
}
if !index.tracks(&did) {
return Ok(());
}
let uri = format!("at://{did}/{}/{}", op.collection, op.rkey);
// Written at the commit's revision, so that neither this nor a
// backfill's read of the repo undoes a later one.
let rev = op.rev.as_deref();
match (op.action.as_str(), op.cid, op.record) {
("delete", _, _) => {
match rev {
Some(rev) => index.delete_rev(&uri, rev)?,
None => index.delete(&uri)?,
};
}
("create" | "update", Some(cid), Some(record)) => {
let put = match rev {
Some(rev) => index.put_rev(&uri, &cid, &record, rev),
None => index.put(&uri, &cid, &record),
};
match put {
Ok(_) => {}
Err(PutError::Refused(why)) => {
tracing::info!(%uri, %why, "jetstream: skipped a record");
}
Err(PutError::Db(e)) => return Err(e),
}
}
_ => {}
}
}
What::Identity { did, handle } => {
if let Some(handle) = handle.filter(|_| index.tracks(&did)) {
index.set_handle(&did, &handle)?;
}
}
What::Account {
did,
active,
status,
} if index.tracks(&did) => match (active, status.as_deref()) {
// Back from being gone, it may have changed while it was: it is
// read whole again.
(true, _) => {
let returned = index.host_status(&did)?.is_some();
index.set_host_status(&did, None)?;
if returned {
index.resync(&did)?;
}
}
(false, Some("deleted")) => {
let purged = index.purge(&did)?;
index.set_host_status(&did, Some("deleted"))?;
tracing::info!(%did, purged, "jetstream: an account was deleted");
}
// Its host says it is not there for now: its records stay,
// unread, for when it is back.
(false, Some(status @ ("deactivated" | "suspended" | "takendown"))) => {
index.set_host_status(&did, Some(status))?;
}
// `desynchronized`, `throttled` and the like say nothing of
// the account itself.
_ => {}
},
What::Info { name, message } => {
tracing::warn!(%name, message = message.unwrap_or_default(), "jetstream says");
}
_ => {}
}
Ok(())
}
crates/grain-serve/src/lib.rs+27−1
View file
@@ -5,9 +5,11 @@
pub mod backfill;
pub mod index;
pub mod jetstream;
pub mod sync;
pub mod xrpc;
use anyhow::bail;
use axum::http::StatusCode;
use axum::{Json, Router, routing::any, routing::get, routing::post};
use grain::env::Ctx;
@@ -27,6 +29,10 @@ pub struct Settings {
/// how to backfill the index. None, as tests and dev run, reads no
/// repos.
pub backfill: Option<backfill::Config>,
/// `GRAIN_JETSTREAM`: the Jetstream to feed the index from, a URL, or
/// `1` for [`jetstream::DEFAULT`]. None, as when it is unset, empty or
/// `0`, follows none, as tests and dev run.
pub jetstream: Option<String>,
}
impl Settings {
@@ -36,6 +42,11 @@ impl Settings {
index: set("GRAIN_INDEX").map(PathBuf::from),
public_url: set("GRAIN_PUBLIC_URL"),
backfill: backfill::Config::from_vars(&get),
jetstream: match set("GRAIN_JETSTREAM").as_deref() {
None | Some("0") => None,
Some("1") => Some(jetstream::DEFAULT.into()),
Some(url) => Some(url.into()),
},
}
}
@@ -51,6 +62,21 @@ impl Settings {
Ok(Some(backfill))
}
/// Starts following the Jetstream these settings name into `app`'s
/// index, on the runtime this is called on; whether there is one.
pub fn follow(&self, app: &App) -> anyhow::Result<bool> {
let Some(url) = &self.jetstream else {
return Ok(false);
};
let Some(index) = &app.index else {
bail!("GRAIN_JETSTREAM needs GRAIN_INDEX");
};
let timing = grain::jetstream::Timing::default();
let follow = jetstream::follow(index.clone(), url.clone(), timing, jetstream::SAVE);
tokio::spawn(follow);
Ok(true)
}
/// The server `ctx` and these settings make: its index opened, and
/// its DID named.
pub fn app(&self, ctx: Ctx) -> anyhow::Result<App> {
@@ -173,7 +199,7 @@ mod tests {
let settings = Settings {
index: Some(dir.path().join("index.db")),
public_url: Some("https://grain.social".into()),
backfill: None,
..Settings::default()
};
let app = settings.app(ctx()).unwrap();
assert!(app.index.is_some());
crates/grain-serve/src/main.rs+4−2
View file
@@ -1,7 +1,8 @@
//! `grain-serve`: listens on `GRAIN_LISTEN`, 127.0.0.1:2600 unless set,
//! keeping the index at `GRAIN_INDEX`, backfilled when `GRAIN_BACKFILL` is
//! set, and taking service tokens for the DID of `GRAIN_PUBLIC_URL` when
//! set ([`grain_serve::Settings`]).
//! set and fed from `GRAIN_JETSTREAM` when that is, and taking service
//! tokens for the DID of `GRAIN_PUBLIC_URL` when set
//! ([`grain_serve::Settings`]).
use anyhow::Context;
use grain::env::{Ctx, Env};
@@ -17,6 +18,7 @@ async fn main() -> anyhow::Result<()> {
if let Some(backfill) = settings.backfill(&app)? {
tokio::spawn(backfill.run());
}
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)
.await
crates/grain-serve/tests/jetstream.rsnew+647−0
View file
647 lines changed: not loaded until asked for.
crates/grain-serve/tests/main.rs+19−1
View file
@@ -1,4 +1,5 @@
//! The `grain-serve` binary: it serves where `GRAIN_LISTEN` says, stops
//! The `grain-serve` binary: it serves where `GRAIN_LISTEN` says, keeps its
//! index where `GRAIN_INDEX` says, stops
//! when told to, and says why when it cannot listen.
use std::process::{Command, Stdio};
@@ -16,8 +17,11 @@ fn free_port() -> u16 {
#[tokio::test]
async fn serves_until_interrupted() {
let addr = format!("127.0.0.1:{}", free_port());
let dir = tempfile::tempdir().unwrap();
let index = dir.path().join("index.db");
let mut child = Command::new(env!("CARGO_BIN_EXE_grain-serve"))
.env("GRAIN_LISTEN", &addr)
.env("GRAIN_INDEX", &index)
.stdout(Stdio::null())
.spawn()
.unwrap();
@@ -31,6 +35,7 @@ async fn serves_until_interrupted() {
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(up, "grain-serve never answered at {addr}");
assert!(index.exists(), "it keeps its index where GRAIN_INDEX says");
// As ctrl-c stops it: gracefully, so it exits 0.
let interrupted = Command::new("kill")
.args(["-INT", &child.id().to_string()])
@@ -52,3 +57,16 @@ fn says_why_it_cannot_listen() {
let said = String::from_utf8_lossy(&out.stderr);
assert!(said.contains(&format!("listening on {addr}")), "{said}");
}
#[test]
fn says_it_cannot_follow_jetstream_without_an_index() {
let out = Command::new(env!("CARGO_BIN_EXE_grain-serve"))
.env("GRAIN_LISTEN", format!("127.0.0.1:{}", free_port()))
.env("GRAIN_JETSTREAM", "1")
.env_remove("GRAIN_INDEX")
.output()
.unwrap();
assert!(!out.status.success());
let said = String::from_utf8_lossy(&out.stderr);
assert!(said.contains("GRAIN_JETSTREAM needs GRAIN_INDEX"), "{said}");
}
crates/grain-serve/Cargo.toml+1−0
View file
@@ -22,4 +22,5 @@ tracing.workspace = true
tracing-subscriber.workspace = true
[dev-dependencies]
tokio-tungstenite.workspace = true
zstd.workspace = true
Cargo.lock+137−4
View file
@@ -293,6 +293,12 @@ dependencies = [
"typenum",
]
[[package]]
name = "data-encoding"
version = "2.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
[[package]]
name = "der"
version = "0.7.10"
@@ -555,6 +561,18 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "getrandom"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
dependencies = [
"cfg-if",
"libc",
"r-efi 5.3.0",
"wasip2",
]
[[package]]
name = "getrandom"
version = "0.4.3"
@@ -564,7 +582,7 @@ dependencies = [
"cfg-if",
"js-sys",
"libc",
"r-efi",
"r-efi 6.0.0",
"rand_core 0.10.1",
"wasm-bindgen",
]
@@ -589,6 +607,7 @@ dependencies = [
"sha2",
"tempfile",
"tokio",
"tokio-tungstenite",
"tracing",
"tracing-subscriber",
"unicode-segmentation",
@@ -610,6 +629,7 @@ dependencies = [
"serde_json",
"tempfile",
"tokio",
"tokio-tungstenite",
"tracing",
"tracing-subscriber",
"zstd",
@@ -772,7 +792,7 @@ dependencies = [
"tokio",
"tokio-rustls",
"tower-service",
"webpki-roots",
"webpki-roots 1.0.11",
]
[[package]]
@@ -1119,6 +1139,15 @@ dependencies = [
"zerovec",
]
[[package]]
name = "ppv-lite86"
version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
dependencies = [
"zerocopy",
]
[[package]]
name = "primeorder"
version = "0.13.6"
@@ -1166,7 +1195,7 @@ dependencies = [
"bytes",
"getrandom 0.4.3",
"lru-slab",
"rand",
"rand 0.10.3",
"rand_pcg",
"ring",
"rustc-hash",
@@ -1202,12 +1231,28 @@ dependencies = [
"proc-macro2",
]
[[package]]
name = "r-efi"
version = "5.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
[[package]]
name = "r-efi"
version = "6.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
[[package]]
name = "rand"
version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41"
dependencies = [
"rand_chacha",
"rand_core 0.9.5",
]
[[package]]
name = "rand"
version = "0.10.3"
@@ -1219,6 +1264,16 @@ dependencies = [
"rand_core 0.10.1",
]
[[package]]
name = "rand_chacha"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
dependencies = [
"ppv-lite86",
"rand_core 0.9.5",
]
[[package]]
name = "rand_core"
version = "0.6.4"
@@ -1228,6 +1283,15 @@ dependencies = [
"getrandom 0.2.17",
]
[[package]]
name = "rand_core"
version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
dependencies = [
"getrandom 0.3.4",
]
[[package]]
name = "rand_core"
version = "0.10.1"
@@ -1296,7 +1360,7 @@ dependencies = [
"wasm-bindgen",
"wasm-bindgen-futures",
"web-sys",
"webpki-roots",
"webpki-roots 1.0.11",
]
[[package]]
@@ -1494,6 +1558,17 @@ dependencies = [
"serde",
]
[[package]]
name = "sha1"
version = "0.10.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8"
dependencies = [
"cfg-if",
"cpufeatures 0.2.17",
"digest",
]
[[package]]
name = "sha2"
version = "0.10.9"
@@ -1739,6 +1814,22 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-tungstenite"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c"
dependencies = [
"futures-util",
"log",
"rustls",
"rustls-pki-types",
"tokio",
"tokio-rustls",
"tungstenite",
"webpki-roots 0.26.11",
]
[[package]]
name = "tokio-util"
version = "0.7.20"
@@ -1871,6 +1962,24 @@ version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
[[package]]
name = "tungstenite"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
dependencies = [
"bytes",
"data-encoding",
"http",
"httparse",
"log",
"rand 0.9.5",
"rustls",
"rustls-pki-types",
"sha1",
"thiserror",
]
[[package]]
name = "typenum"
version = "1.20.1"
@@ -1955,6 +2064,15 @@ version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "wasip2"
version = "1.0.4+wasi-0.2.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487"
dependencies = [
"wit-bindgen",
]
[[package]]
name = "wasm-bindgen"
version = "0.2.130"
@@ -2030,6 +2148,15 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "webpki-roots"
version = "0.26.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
dependencies = [
"webpki-roots 1.0.11",
]
[[package]]
name = "webpki-roots"
version = "1.0.11"
@@ -2127,6 +2254,12 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
[[package]]
name = "wit-bindgen"
version = "0.57.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
[[package]]
name = "writeable"
version = "0.6.4"
Cargo.toml+1−0
View file