grain.social/grain
photography-focused photo sharing platform
grain.social
1 collaborator
- 1 collaborator
0
2
The Justfile runs grain-serve and the web, as knotless's does
Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.
.knotless/workflows/check.yml+6−0
| @@ -1,8 +1,14 @@ | ||
# Formatting, lint and types, as `npm run check` runs them: vp check, then | ||
# svelte-check. The packages are kept between runs and installed again only | ||
# when the lockfile changes. | ||
# | ||
# The Node side's: at cutover rust.yml is renamed over it, and what of it | ||
# checks web/ moves to a web.yml of its own, as knotless's does. | ||
image: node:25-slim | ||
runners: [runner.knotless.dev] | ||
# A pre-flight: the merge queue takes a pull into a batch only once this | ||
# has passed at its head, so one that does not build fails no batch. | ||
queue: first | ||
on: | ||
push: { refs: [refs/heads/main] } | ||
pull: { base: [refs/heads/main, refs/heads/feature/*] } | ||
.knotless/workflows/deploy.yml+25−5
| @@ -1,6 +1,12 @@ | ||
# Deploys main to the Hetzner host once this commit's test and check have | ||
# passed. The appview image is built from this checkout on the host, over | ||
# SSH, as a candidate, grain:build-<commit>, while the checks run; once | ||
# Deploys main to the Hetzner host once this commit's checks all pass: | ||
# check and test, the Node app's, rust and harness, the Rust | ||
# workspace's, and lexicons, so that no part can fail and still deploy. | ||
# Until cutover the Node app is what deploys; at cutover rust.yml and | ||
# harness.yml become check.yml and test.yml, and the list below is | ||
# `check test lexicons`. | ||
# | ||
# The appview image is built from this checkout on the host, over SSH, as | ||
# a candidate, grain:build-<commit>, while the checks run; once | ||
# they pass it becomes grain:current and the grain service is recreated | ||
# from it, leaving the rest of the stack as it is. grain-deploy on the | ||
# host rebuilds the whole stack by hand. | ||
| @@ -38,12 +44,26 @@ steps: | ||
timeout: 30 | ||
run: | | ||
DOCKER_HOST="ssh://$GRAIN_HOST" docker build -t "grain:build-$KNOTLESS_RUN_SHA" . | ||
- name: checks | ||
- name: checks done | ||
timeout: 30 | ||
run: | | ||
# 25 minutes over all five, and never a failure: the next step | ||
# judges. Its checks can wait for a job behind other runs, so it | ||
# waits for them over two steps, each under the runner's 30 minutes | ||
# a step, as knotless's deploy does. | ||
repo="knotless://$KNOTLESS_RUN_REPO" | ||
end=$(($(date +%s) + 1500)) | ||
for wf in check test rust harness lexicons; do | ||
left=$((end - $(date +%s))) | ||
[ "$left" -gt 0 ] || break | ||
knotless checks "$repo" --ref "$KNOTLESS_RUN_REF" --commit "$KNOTLESS_RUN_SHA" --workflow "$wf" --wait "$left" || true | ||
done | ||
- name: checks pass | ||
timeout: 30 | ||
run: | | ||
repo="knotless://$KNOTLESS_RUN_REPO" | ||
end=$(($(date +%s) + 1500)) | ||
for wf in test check; do | ||
for wf in check test rust harness lexicons; do | ||
left=$((end - $(date +%s))) | ||
[ "$left" -gt 0 ] || left=1 | ||
out=$(knotless checks "$repo" --ref "$KNOTLESS_RUN_REF" --commit "$KNOTLESS_RUN_SHA" --workflow "$wf" --wait "$left") | ||
.knotless/workflows/harness.yml+7−0
| @@ -13,8 +13,15 @@ | ||
# Node, the harness's packages, the alpha's build, cargo-llvm-cov and the | ||
# build are kept between runs; each is made again only when what it is | ||
# made from changes. | ||
# | ||
# At cutover this becomes test.yml, as knotless's is, over the Node one, | ||
# and rust.yml check.yml (see there). | ||
image: rust:1.98 | ||
runners: [runner.knotless.dev] | ||
# A pre-flight, as check is: the merge queue takes a pull into a batch only | ||
# once this has passed at its head, so one whose tests fail on their own | ||
# costs no batch and no halving. | ||
queue: first | ||
on: | ||
push: { refs: [refs/heads/main] } | ||
pull: { base: [refs/heads/main, refs/heads/feature/*] } | ||
.knotless/workflows/lexicons.ymlnew+30−0
| @@ -0,0 +1,30 @@ | ||
# The lexicons, as `just lex` checks them: goat's lint, what it does not | ||
# see, and that they only grow from main's (harness/lexicons.mjs), over | ||
# grain's own under lexicons/social/grain, less what the old server alone | ||
# serves. goat is kept between runs. | ||
# | ||
# The XRPC answers are held to the lexicons by harness.yml, over what | ||
# grain-serve's tests received. | ||
image: node:24.22.1 | ||
runners: [runner.knotless.dev] | ||
# Every push to main, and pulls into it or a feature/ branch, at their | ||
# heads merged with their base, that change the lexicons, their check or | ||
# this file. A pull changing none passes at once, noted as skipped. | ||
on: | ||
push: { refs: [refs/heads/main] } | ||
pull: | ||
base: [refs/heads/main, refs/heads/feature/*] | ||
paths: [lexicons/**, harness/lexicons.mjs, .knotless/workflows/lexicons.yml] | ||
cache: [/opt/goat] | ||
steps: | ||
- name: goat | ||
run: | | ||
mkdir -p /opt/goat | ||
[ -x /opt/goat/goat ] || curl -fsSL https://github.com/bluesky-social/goat/releases/download/v0.2.6/goat_Linux_x86_64.tar.gz | tar -xz -C /opt/goat goat | ||
/opt/goat/goat --version | ||
- name: lexicons | ||
run: | | ||
# The checkout is another user's: git reads where it left main only | ||
# once told it may. | ||
git config --global --add safe.directory "$PWD" | ||
PATH=/opt/goat:$PATH node harness/lexicons.mjs | ||
.knotless/workflows/publish-lexicons.yml+4−3
| @@ -1,5 +1,6 @@ | ||
# Publishes grain's lexicons as atproto resolves them, once main's test | ||
# and check pass at a commit that changes them (deploy/publish-lexicons.mjs): | ||
# Publishes grain's lexicons as atproto resolves them, once main's checks | ||
# all pass (check, test, rust, harness and lexicons, as deploy.yml waits on) at a | ||
# commit that changes them (deploy/publish-lexicons.mjs): | ||
# a com.atproto.lexicon.schema record in grain.social's repo for each | ||
# social.grain.* lexicon, and the `_lexicon.<authority>` TXT records in the | ||
# grain.social zone naming that account. What is new or changed is | ||
| @@ -42,7 +43,7 @@ steps: | ||
curl -fsSL https://knotless.dev/install.sh | KNOTLESS_INSTALL_DIR=/usr/local/bin sh | ||
repo="knotless://$KNOTLESS_RUN_REPO" | ||
end=$(($(date +%s) + 1500)) | ||
for wf in test check; do | ||
for wf in check test rust harness lexicons; do | ||
left=$((end - $(date +%s))) | ||
[ "$left" -gt 0 ] || left=1 | ||
out=$(knotless checks "$repo" --ref "$KNOTLESS_RUN_REF" --commit "$KNOTLESS_RUN_SHA" --workflow "$wf" --wait "$left") | ||
.knotless/workflows/rust.yml+8−0
| @@ -2,10 +2,18 @@ | ||
# lint` runs them; harness.yml runs the tests. Kept apart from check.yml | ||
# and test.yml, which run the Node side of the repository. | ||
# | ||
# At cutover this becomes check.yml, as knotless's is, and harness.yml | ||
# test.yml: the Node check.yml and test.yml are deleted, this file is | ||
# renamed over check.yml, and the lists of checks deploy.yml and | ||
# publish-lexicons.yml wait on go back to `check test` (and lexicons). | ||
# | ||
# The build, the registry and the toolchain's components are kept between | ||
# runs, so a push builds only what it changed. | ||
image: rust:1.98 | ||
runners: [runner.knotless.dev] | ||
# A pre-flight: the merge queue takes a pull into a batch only once this | ||
# has passed at its head, so one that does not build fails no batch. | ||
queue: first | ||
on: | ||
push: { refs: [refs/heads/main] } | ||
pull: { base: [refs/heads/main, refs/heads/feature/*] } | ||
.knotless/workflows/test.yml+6−0
| @@ -2,8 +2,14 @@ | ||
# themselves without SPACES_LIVE=1, and the Playwright tests in test/browser | ||
# are not run here. The packages are kept between runs and installed again | ||
# only when the lockfile changes. | ||
# | ||
# The Node side's: at cutover harness.yml is renamed over it. | ||
image: node:25-slim | ||
runners: [runner.knotless.dev] | ||
# A pre-flight, as check is: the merge queue takes a pull into a batch only | ||
# once this has passed at its head, so one whose tests fail on their own | ||
# costs no batch and no halving. | ||
queue: first | ||
on: | ||
push: { refs: [refs/heads/main] } | ||
pull: { base: [refs/heads/main, refs/heads/feature/*] } | ||
crates/grain/src/telemetry.rs+548−14
562 lines changed: not loaded until asked for.
crates/grain/Cargo.toml+1−1
| @@ -23,10 +23,10 @@ sha2.workspace = true | ||
tokio.workspace = true | ||
tokio-tungstenite.workspace = true | ||
tracing.workspace = true | ||
tracing-subscriber.workspace = true | ||
unicode-segmentation.workspace = true | ||
zstd.workspace = true | ||
| ||
[dev-dependencies] | ||
axum.workspace = true | ||
tempfile.workspace = true | ||
tracing-subscriber.workspace = true | ||
crates/grain-serve/src/index/mod.rs+34−1
| @@ -48,6 +48,10 @@ pub struct Index { | ||
/// Where the same records are told for the classifiers | ||
/// ([`Index::classify_to`]). | ||
classifying: std::sync::OnceLock<tokio::sync::mpsc::Sender<String>>, | ||
/// Its file, unless it is held in memory. | ||
path: Option<std::path::PathBuf>, | ||
/// The feeds that keep it current, as `/_status` tells of them. | ||
pub feeds: crate::status::Feeds, | ||
} | ||
| ||
/// What putting a record did. | ||
| @@ -190,7 +194,33 @@ impl Index { | ||
/// The index at `path`, made or migrated to this version's tables. | ||
pub fn open(path: &Path) -> anyhow::Result<Index> { | ||
let db = Connection::open(path).with_context(|| format!("opening {}", path.display()))?; | ||
Index::with(db, schema::MIGRATIONS) | ||
let mut index = Index::with(db, schema::MIGRATIONS)?; | ||
index.path = Some(path.to_path_buf()); | ||
Ok(index) | ||
} | ||
| ||
/// The version of the tables in the index at `path`, read without | ||
/// changing it; none where there is no file yet. | ||
pub fn version_at(path: &Path) -> anyhow::Result<Option<usize>> { | ||
if !path.exists() { | ||
return Ok(None); | ||
} | ||
let flags = rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY; | ||
let db = Connection::open_with_flags(path, flags) | ||
.with_context(|| format!("opening {}", path.display()))?; | ||
let at: i64 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?; | ||
Ok(Some(at as usize)) | ||
} | ||
| ||
/// The index's file, unless it is held in memory. | ||
pub fn path(&self) -> Option<&Path> { | ||
self.path.as_deref() | ||
} | ||
| ||
/// That the index answers a read. | ||
pub fn answers(&self) -> rusqlite::Result<()> { | ||
let sql = "SELECT EXISTS (SELECT 1 FROM repo)"; | ||
self.read(|db| db.query_row(sql, [], |r| r.get::<_, bool>(0)).map(drop)) | ||
} | ||
| ||
/// An index held in memory, for tests. | ||
| @@ -200,6 +230,7 @@ impl Index { | ||
| ||
fn with(mut db: Connection, steps: &[&str]) -> anyhow::Result<Index> { | ||
db.busy_timeout(std::time::Duration::from_secs(5))?; | ||
crate::telemetry::watch(&db); | ||
db.execute_batch("PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL;")?; | ||
migrate(&mut db, steps)?; | ||
let tracked = { | ||
| @@ -213,6 +244,8 @@ impl Index { | ||
tracked: Mutex::new(tracked), | ||
arrivals: std::sync::OnceLock::new(), | ||
classifying: std::sync::OnceLock::new(), | ||
path: None, | ||
feeds: crate::status::Feeds::default(), | ||
}) | ||
} | ||
| ||
crates/grain-serve/src/jetstream.rs+6−0
| @@ -29,6 +29,9 @@ use std::time::{Duration, Instant}; | ||
/// The Jetstream grain follows unless told another. | ||
pub const DEFAULT: &str = "wss://jetstream.us-west.bsky.network"; | ||
| ||
/// The feed's name, as `/_status` tells of it. | ||
pub const NAME: &str = "jetstream"; | ||
| ||
/// How often where the stream was read to is saved, at most. | ||
pub const SAVE: Duration = Duration::from_secs(5); | ||
| ||
| @@ -42,6 +45,7 @@ pub fn source(url: &str) -> String { | ||
/// 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(); | ||
index.feeds.follows(NAME, &url); | ||
let feed = Feed { | ||
source: source(&url), | ||
index, | ||
| @@ -87,10 +91,12 @@ impl Follower for Feed { | ||
| ||
fn connected(&mut self, up: bool) { | ||
tracing::info!(up, source = %self.source, "jetstream: connecting"); | ||
self.index.feeds.connected(NAME, up); | ||
} | ||
| ||
fn event(&mut self, event: Event) { | ||
self.seq = event.seq.or(self.seq); | ||
self.index.feeds.heard(NAME, event.seq); | ||
logged(apply(&self.index, event)); | ||
if self.saved.is_none_or(|at| at.elapsed() >= self.save) { | ||
logged(self.save()); | ||