chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
A backfill asks a directory which accounts to read, and reads them from a mirror as STAR-lite
crates/knotless-serve/fixtures/repo.carnew+0−0
A binary file, not shown.
crates/knotless-serve/fixtures/repo.jsonnew+6−0
| @@ -0,0 +1,6 @@ | ||
{ | ||
"did": "did:plc:archivefixture00000000000", | ||
"key": "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme", | ||
"rev": "3mzcpijkfus2c", | ||
"records": 60 | ||
} | ||
crates/knotless-serve/fixtures/repo.starnew+0−0
A binary file, not shown.
crates/knotless-serve/src/archive.rs+474−66
540 lines changed: not loaded until asked for.
crates/knotless-serve/src/lib.rs+79−19
| @@ -663,8 +663,13 @@ pub async fn index_op( | ||
} | ||
| ||
/// Backfills from `service`, saying how it went. | ||
pub async fn backfilled(ctx: &Ctx, index: &index::Index, service: &str) -> String { | ||
match backfill(ctx, index, service).await { | ||
pub async fn backfilled( | ||
ctx: &Ctx, | ||
index: &index::Index, | ||
service: &str, | ||
sources: &Sources, | ||
) -> String { | ||
match backfill(ctx, index, service, sources).await { | ||
Ok(f) => format!( | ||
"backfilled {} records of {} accounts from {service}, {} unchanged, {} failed", | ||
f.records, f.read, f.unchanged, f.failed | ||
| @@ -699,14 +704,29 @@ fn indexed_collections() -> impl Iterator<Item = &'static str> { | ||
/// The accounts on `xrpc`'s service to index: from a relay, those it lists | ||
/// as holding an indexed collection, a few among millions; from a PDS, | ||
/// which lists accounts only all at once, every one. | ||
async fn accounts(xrpc: &knotless::xrpc::Xrpc) -> Result<Vec<String>> { | ||
/// | ||
/// Each of `directories` is asked first, in turn, and the first to answer | ||
/// for every collection is believed. | ||
async fn accounts(ctx: &Ctx, service: &str, directories: &[String]) -> Result<Vec<String>> { | ||
for directory in directories { | ||
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), directory); | ||
if let Ok(dids) = holders(&xrpc).await { | ||
return Ok(dids); | ||
} | ||
} | ||
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), service); | ||
match holders(&xrpc).await { | ||
Ok(dids) => Ok(dids), | ||
Err(_) => pages(&xrpc, "com.atproto.sync.listRepos", None).await, | ||
} | ||
} | ||
| ||
/// The accounts `xrpc`'s service lists as holding an indexed collection. | ||
async fn holders(xrpc: &knotless::xrpc::Xrpc) -> Result<Vec<String>> { | ||
let mut dids = std::collections::BTreeSet::new(); | ||
for collection in indexed_collections() { | ||
let by = "com.atproto.sync.listReposByCollection"; | ||
match pages(xrpc, by, Some(collection)).await { | ||
Ok(found) => dids.extend(found), | ||
Err(_) => return pages(xrpc, "com.atproto.sync.listRepos", None).await, | ||
} | ||
dids.extend(pages(xrpc, by, Some(collection)).await?); | ||
} | ||
Ok(dids.into_iter().collect()) | ||
} | ||
| @@ -743,6 +763,18 @@ async fn pages( | ||
Ok(dids) | ||
} | ||
| ||
/// Where a backfill finds the accounts to read and reads them, besides the | ||
/// PDS or relay it is from and each account's PDS. | ||
#[derive(Clone, Debug, Default, PartialEq)] | ||
pub struct Sources { | ||
/// Directories asked, in turn, which accounts hold knotless records | ||
/// (`com.atproto.sync.listReposByCollection`), before the relay is. | ||
pub directories: Vec<String>, | ||
/// A mirror each account's repository is read from first, as | ||
/// STAR-lite or a CAR ([`archive::fetch`]). | ||
pub mirror: Option<String>, | ||
} | ||
| ||
/// What a backfill did: the records it indexed, from the accounts it | ||
/// read; the accounts it found unchanged since it last read them, or put | ||
/// off as one failed not long ago; and those it failed to read. | ||
| @@ -766,12 +798,18 @@ const READERS: usize = 4; | ||
/// ([`index::Index::put_as_of`]). An account that fails is tried again | ||
/// by a later backfill, a minute later for each time it failed, up to an | ||
/// hour, and does not stop the rest. | ||
pub async fn backfill(ctx: &Ctx, index: &index::Index, service: &str) -> Result<Filled> { | ||
pub async fn backfill( | ||
ctx: &Ctx, | ||
index: &index::Index, | ||
service: &str, | ||
sources: &Sources, | ||
) -> Result<Filled> { | ||
use futures::StreamExt; | ||
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), service); | ||
let reads = futures::stream::iter(accounts(&xrpc).await?) | ||
let found = accounts(ctx, service, &sources.directories).await?; | ||
let mirror = sources.mirror.as_deref(); | ||
let reads = futures::stream::iter(found) | ||
.map(|did| async move { | ||
let read = read_account(ctx, index, &did).await; | ||
let read = read_account(ctx, index, &did, mirror).await; | ||
(did, read) | ||
}) | ||
.buffer_unordered(READERS); | ||
| @@ -794,9 +832,15 @@ pub async fn backfill(ctx: &Ctx, index: &index::Index, service: &str) -> Result< | ||
Ok(filled) | ||
} | ||
| ||
/// Reads `did` into the index, and how many records it held; `None` when | ||
/// it is as last read, or a failure put it off until later. | ||
async fn read_account(ctx: &Ctx, index: &index::Index, did: &str) -> Result<Option<usize>> { | ||
/// Reads `did` into the index, from `mirror` when it has it, and how many | ||
/// records it held; `None` when it is as last read, or a failure put it off | ||
/// until later. | ||
async fn read_account( | ||
ctx: &Ctx, | ||
index: &index::Index, | ||
did: &str, | ||
mirror: Option<&str>, | ||
) -> Result<Option<usize>> { | ||
#[derive(serde::Deserialize)] | ||
struct Latest { | ||
rev: String, | ||
| @@ -813,7 +857,7 @@ async fn read_account(ctx: &Ctx, index: &index::Index, did: &str) -> Result<Opti | ||
return Ok(None); | ||
} | ||
let kept = |c: &str| indexed_collections().any(|i| i == c); | ||
let read = archive::fetch(ctx, did, &id.pds, archive::MAX, kept).await?; | ||
let read = archive::fetch(ctx, did, &id.pds, mirror, archive::MAX, &kept).await?; | ||
for r in &read.records { | ||
index.put_as_of(did, &r.collection, &r.rkey, &r.value, Some(&read.rev))?; | ||
} | ||
| @@ -837,11 +881,17 @@ const SUMMARY: Duration = Duration::from_secs(60); | ||
/// 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) { | ||
pub async fn sweep( | ||
ctx: Ctx, | ||
index: Arc<index::Index>, | ||
relay: String, | ||
sources: Sources, | ||
every: Duration, | ||
) { | ||
loop { | ||
tokio::time::sleep(every).await; | ||
let _ = index.lapse(); | ||
telemetry::sweep(&ctx, &index, &relay).await; | ||
telemetry::sweep(&ctx, &index, &relay, &sources).await; | ||
} | ||
} | ||
| ||
| @@ -1045,6 +1095,9 @@ pub struct Options { | ||
/// the index starts empty, so a new version of its tables loses | ||
/// nothing older than what Jetstream replays. | ||
pub relay: Option<String>, | ||
/// Where a backfill finds accounts and reads them first: directories, | ||
/// and a mirror. | ||
pub sources: Sources, | ||
/// Labelers, by DID, whose takedowns hold here too. | ||
pub labelers: Vec<String>, | ||
/// The runner, by DID, a new public repository names unless its | ||
| @@ -1096,6 +1149,7 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> { | ||
index: None, | ||
backfill: None, | ||
relay: None, | ||
sources: Sources::default(), | ||
labelers: Vec::new(), | ||
default_runner: None, | ||
client_minimum: None, | ||
| @@ -1132,6 +1186,11 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> { | ||
"--index" => opts.index = Some(value()?.into()), | ||
"--backfill" => opts.backfill = Some(value()?).filter(|v| !v.is_empty()), | ||
"--relay" => opts.relay = Some(value()?), | ||
"--directory" => opts | ||
.sources | ||
.directories | ||
.extend(Some(value()?).filter(|v| !v.is_empty())), | ||
"--mirror" => opts.sources.mirror = Some(value()?).filter(|v| !v.is_empty()), | ||
"--labeler" => opts.labelers.push(value()?), | ||
"--default-runner" => { | ||
let did = value()?; | ||
| @@ -1261,11 +1320,12 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> { | ||
let backfill = opts.backfill.or(opts.relay.clone().filter(|_| rebuilt)); | ||
if let (Some(service), Some(index)) = (backfill, app.index.clone()) { | ||
let ctx = app.ctx.clone(); | ||
tokio::spawn(async move { telemetry::sweep(&ctx, &index, &service).await }); | ||
let sources = opts.sources.clone(); | ||
tokio::spawn(async move { telemetry::sweep(&ctx, &index, &service, &sources).await }); | ||
} | ||
// And again now and then, for what the feeds missed. | ||
if let (Some(relay), Some(index)) = (opts.relay, app.index.clone()) { | ||
tokio::spawn(sweep(app.ctx.clone(), index, relay, SWEEP)); | ||
tokio::spawn(sweep(app.ctx.clone(), index, relay, opts.sources, SWEEP)); | ||
} | ||
for did in opts.labelers { | ||
tokio::spawn(moderation::follow(app.clone(), did)); | ||
crates/knotless-serve/src/telemetry.rs+7−2
| @@ -378,14 +378,19 @@ pub fn opened(index: Option<&crate::index::Index>, took: Duration) { | ||
| ||
/// Fills the index from `service`, saying when it starts and how it | ||
/// ended. | ||
pub async fn sweep(ctx: &knotless::env::Ctx, index: &crate::index::Index, service: &str) { | ||
pub async fn sweep( | ||
ctx: &knotless::env::Ctx, | ||
index: &crate::index::Index, | ||
service: &str, | ||
sources: &crate::Sources, | ||
) { | ||
let started = Instant::now(); | ||
let source = Fields { | ||
source: Some(service), | ||
..Fields::default() | ||
}; | ||
telemetry::note(Level::INFO, "index sweep started", &source); | ||
let said = crate::backfilled(ctx, index, service).await; | ||
let said = crate::backfilled(ctx, index, service, sources).await; | ||
let fields = Fields { | ||
reason: Some(&said), | ||
duration_ms: Some(telemetry::millis(started.elapsed())), | ||
crates/knotless-serve/tests/forge.rs+77−4
| @@ -111,6 +111,33 @@ fn listing(av: &Appview, dids: &[&str]) -> String { | ||
}) | ||
} | ||
| ||
/// A mirror, as hubble is one, that answers `getRepo` with the CAR the | ||
/// account's PDS has, and counts what it was asked. | ||
fn mirror(av: &Appview) -> (String, Arc<std::sync::atomic::AtomicUsize>) { | ||
let asked = Arc::new(std::sync::atomic::AtomicUsize::new(0)); | ||
let count = asked.clone(); | ||
let get_repo = axum::routing::get( | ||
move |q: axum::extract::Query<std::collections::HashMap<String, String>>| { | ||
count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); | ||
async move { | ||
let did = q.get("did").cloned().unwrap_or_default(); | ||
let url = format!("{}/xrpc/com.atproto.sync.getRepo", net().pds); | ||
let res = reqwest::Client::new().get(url).query(&[("did", did)]); | ||
let car = res.send().await.unwrap().bytes().await.unwrap(); | ||
([("content-type", "application/vnd.ipld.car")], car.to_vec()) | ||
} | ||
}, | ||
); | ||
let router = axum::Router::new().route("/xrpc/com.atproto.sync.getRepo", get_repo); | ||
let url = av.rt.block_on(async move { | ||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); | ||
let url = format!("http://{}", listener.local_addr().unwrap()); | ||
tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); | ||
url | ||
}); | ||
(url, asked) | ||
} | ||
| ||
#[test] | ||
fn forks_pulls_reviews_and_merges() { | ||
let av = appview(); | ||
| @@ -479,15 +506,17 @@ fn forks_pulls_reviews_and_merges() { | ||
let s = Sandbox::new(); | ||
let index = Index::open(&s.path("i.db")).unwrap(); | ||
let pds = listing(&av, &[&owner.account.did, &contrib.account.did]); | ||
let none = knotless_serve::Sources::default(); | ||
let n = av | ||
.rt | ||
.block_on(knotless_serve::backfill(&ctx(&s), &index, &pds)) | ||
.block_on(knotless_serve::backfill(&ctx(&s), &index, &pds, &none)) | ||
.unwrap(); | ||
assert!(n.records >= 10 && n.read == 2, "{n:?}"); | ||
let report = av.rt.block_on(knotless_serve::backfilled( | ||
&ctx(&s), | ||
&index, | ||
"http://127.0.0.1:9", | ||
&none, | ||
)); | ||
assert!( | ||
report.starts_with("backfilling from http://127.0.0.1:9: "), | ||
| @@ -496,7 +525,7 @@ fn forks_pulls_reviews_and_merges() { | ||
// Read again with nothing new, each account is passed over. | ||
let report = av | ||
.rt | ||
.block_on(knotless_serve::backfilled(&ctx(&s), &index, &pds)); | ||
.block_on(knotless_serve::backfilled(&ctx(&s), &index, &pds, &none)); | ||
assert!(report.ends_with(", 2 unchanged, 0 failed"), "{report}"); | ||
let did = owner.account.did.as_str(); | ||
let backfilled = index.forge(did, &key).unwrap().len(); | ||
| @@ -528,7 +557,7 @@ fn forks_pulls_reviews_and_merges() { | ||
assert!(fresh.rebuilt); | ||
let filled = av | ||
.rt | ||
.block_on(knotless_serve::backfill(&ctx(&s), &fresh, &relay)) | ||
.block_on(knotless_serve::backfill(&ctx(&s), &fresh, &relay, &none)) | ||
.unwrap(); | ||
// One it cannot read is tried again later, and does not stop the rest. | ||
assert_eq!((filled.read, filled.failed), (1, 1), "{filled:?}"); | ||
| @@ -536,7 +565,7 @@ fn forks_pulls_reviews_and_merges() { | ||
assert_eq!((nobody.rev, nobody.tries), (None, 1)); | ||
let again = av | ||
.rt | ||
.block_on(knotless_serve::backfill(&ctx(&s), &fresh, &relay)) | ||
.block_on(knotless_serve::backfill(&ctx(&s), &fresh, &relay, &none)) | ||
.unwrap(); | ||
assert_eq!((again.unchanged, again.failed), (2, 0), "{again:?}"); | ||
// Only the owner's: what others wrote, the relay here does not name. | ||
| @@ -545,12 +574,56 @@ fn forks_pulls_reviews_and_merges() { | ||
assert!(owners > 0 && owners < backfilled, "{owners}"); | ||
// Opened again, it is the same version, and kept. | ||
assert!(!Index::open(&s.path("relay.db")).unwrap().rebuilt); | ||
// A directory that answers is asked before the relay, one that does | ||
// not passed by, and each repository read from the mirror. | ||
let (mirror, asked) = mirror(&av); | ||
let sources = knotless_serve::Sources { | ||
directories: vec!["http://127.0.0.1:9".into(), relay.clone()], | ||
mirror: Some(mirror), | ||
}; | ||
let mirrored = Index::open(&s.path("mirrored.db")).unwrap(); | ||
let filled = av | ||
.rt | ||
.block_on(knotless_serve::backfill( | ||
&ctx(&s), | ||
&mirrored, | ||
"http://127.0.0.1:9", | ||
&sources, | ||
)) | ||
.unwrap(); | ||
assert_eq!((filled.read, filled.failed), (1, 1), "{filled:?}"); | ||
assert_eq!(asked.load(std::sync::atomic::Ordering::Relaxed), 1); | ||
assert_eq!(mirrored.forge(did, &key).unwrap().len(), owners); | ||
// A mirror that cannot be reached, or does not hold the repository, | ||
// leaves each read to its PDS. | ||
for (i, gone) in ["http://127.0.0.1:9", relay.as_str()] | ||
.into_iter() | ||
.enumerate() | ||
{ | ||
let gone = knotless_serve::Sources { | ||
directories: vec![relay.clone()], | ||
mirror: Some(gone.into()), | ||
}; | ||
let unmirrored = Index::open(&s.path(&format!("unmirrored{i}.db"))).unwrap(); | ||
let filled = av | ||
.rt | ||
.block_on(knotless_serve::backfill( | ||
&ctx(&s), | ||
&unmirrored, | ||
"http://127.0.0.1:9", | ||
&gone, | ||
)) | ||
.unwrap(); | ||
assert_eq!(filled.read, 1, "{filled:?}"); | ||
assert_eq!(unmirrored.forge(did, &key).unwrap().len(), owners); | ||
} | ||
// A sweep fills it again, now and then, with what a feed missed. | ||
let swept = std::sync::Arc::new(Index::open(&s.path("swept.db")).unwrap()); | ||
let sweep = av.rt.spawn(knotless_serve::sweep( | ||
ctx(&s), | ||
swept.clone(), | ||
relay.clone(), | ||
none.clone(), | ||
std::time::Duration::from_millis(50), | ||
)); | ||
wait_until(|| swept.repo(&uri).unwrap().is_some()); | ||
crates/knotless-serve/tests/serve.rs+15−0
| @@ -516,6 +516,21 @@ fn arguments() { | ||
assert_eq!((o.firehose, o.jetstream, o.backfill), (None, None, None)); | ||
let o = args(&["--backfill", "https://r"]).unwrap(); | ||
assert_eq!(o.backfill.as_deref(), Some("https://r")); | ||
// Directories in turn, and a mirror; empty, none. | ||
let o = args(&[ | ||
"--directory", | ||
"https://a", | ||
"--directory", | ||
"", | ||
"--directory", | ||
"https://b", | ||
"--mirror", | ||
"https://m", | ||
]) | ||
.unwrap(); | ||
assert_eq!(o.sources.directories, ["https://a", "https://b"]); | ||
assert_eq!(o.sources.mirror.as_deref(), Some("https://m")); | ||
assert_eq!(args(&["--mirror", ""]).unwrap().sources.mirror, None); | ||
let gone = args(&["--private"]).err(); | ||
assert_eq!(gone.as_deref(), Some(knotless_serve::PRIVATE_GONE)); | ||
let o = args(&["--public-url", "https://knot.example"]).unwrap(); | ||
deploy/docker-compose.yml+8−0
| @@ -52,6 +52,14 @@ services: | ||
# records. | ||
- --relay | ||
- ${KNOTLESS_RELAY:-https://relay1.us-east.bsky.network} | ||
# Which accounts hold knotless records is asked of microcosm's | ||
# lightrail before the relay, and each account's repository is read | ||
# from its hubble mirror before the account's PDS. Either set empty | ||
# is not asked. | ||
- --directory | ||
- ${KNOTLESS_DIRECTORY-https://lightrail.microcosm.blue} | ||
- --mirror | ||
- ${KNOTLESS_MIRROR-https://hubble.microcosm.blue} | ||
- --web | ||
- http://pages:2591 | ||
# The client's builds, as the repository's releases of its | ||
docs/design/serve.md+26−15
| @@ -463,21 +463,32 @@ rebuilt index otherwise starts empty, and a feed only brings what happens | ||
next, so it needs a backfill. | ||
| ||
A backfill reads the knotless records of every account that holds any. A relay | ||
can name those accounts by collection; a PDS lists all of its own. `--backfill` | ||
runs one at startup. `--relay` runs one when the index was rebuilt, then sweeps | ||
every 10 minutes, since a record whose event a feed never delivered would | ||
otherwise be missing for good. | ||
| ||
Each account is read whole, as one CAR from its PDS (`com.atproto.sync.getRepo`), | ||
a few accounts at a time, and up to 256 MiB each. Nothing in it is believed | ||
until it checks out: the commit must be the account's and signed by the key its | ||
DID document names, and every block on the way from the commit to a record must | ||
hash to its CID. Only the indexed collections' records are kept. An account is | ||
first asked for its latest commit (`com.atproto.sync.getLatestCommit`), and one | ||
whose commit is the one last read is passed over, so a sweep reads only the | ||
accounts that changed. An account that fails is logged as `account not | ||
backfilled` and tried again by a later backfill, a minute later for each time it | ||
has failed, up to an hour. It does not stop the others. | ||
can name those accounts by collection; a PDS lists all of its own. A collection | ||
directory, such as microcosm's lightrail, names them as a relay does, and each | ||
given with `--directory` is asked first, in turn, the relay or PDS only when | ||
none answers. `--backfill` runs one at startup. `--relay` runs one when the | ||
index was rebuilt, then sweeps every 10 minutes, since a record whose event a | ||
feed never delivered would otherwise be missing for good. | ||
| ||
Each account is read whole, a few accounts at a time, and up to 256 MiB each. | ||
With `--mirror`, the repository is asked of that mirror first, such as | ||
microcosm's hubble, as STAR-lite or else a CAR. STAR-lite is every key and | ||
record in key order, with the commit in its header and no tree, so it is read | ||
as it streams: the MST's root is rebuilt from the keys, holding one node a layer, | ||
and must be the one the commit signed. A mirror that cannot be reached, or does | ||
not have the repository, leaves the read to the account's PDS, as one CAR | ||
(`com.atproto.sync.getRepo`), held whole to walk its tree. A mirror that asks to | ||
be asked less fails the read, for a later backfill to try again. | ||
| ||
Nothing read is believed until it checks out: the commit must be the account's | ||
and signed by the key its DID document names, and the records must be the ones | ||
its MST holds, block by block for a CAR, or as the rebuilt root for STAR-lite. | ||
Only the indexed collections' records are kept. An account is first asked for | ||
its latest commit (`com.atproto.sync.getLatestCommit`), and one whose commit is | ||
the one last read is passed over, so a sweep reads only the accounts that | ||
changed. An account that fails is logged as `account not backfilled` and tried | ||
again by a later backfill, a minute later for each time it has failed, up to an | ||
hour. It does not stop the others. | ||
| ||
A backfill reads while the feed goes on, so it must not put back what the feed | ||
has since changed. Each write the index takes from a feed keeps the revision of | ||
docs/reference/cli.md+2−0
| @@ -372,6 +372,8 @@ has neither. | ||
| `--index <file>` | | keep an appview index in this SQLite file | | ||
| `--backfill <https://pds or relay>` | | at start, index every account on a PDS, or those holding knotless records on a relay | | ||
| `--relay <https://relay>` | | backfill from this relay when the index was rebuilt (a new file, or another build's tables), and every ten minutes after | | ||
| `--directory <https://host>` | | ask this collection directory (`com.atproto.sync.listReposByCollection`, as lightrail answers it) which accounts hold knotless records before the relay or PDS a backfill is from; repeatable, asked in turn; empty, none | | ||
| `--mirror <https://host>` | | read each account's repository from this mirror (`com.atproto.sync.getRepo`, as hubble answers it, STAR-lite or a CAR) before the account's PDS; empty, none | | ||
| `--public-url <https://host>` | | the URL the server is reached at; makes it a gateway, as `did:web` of the host, and serves MCP at `/mcp` | | ||
| `--labeler <did>` | | honour this labeler's takedowns; repeatable | | ||
| `--default-runner <did>` | | the runner a new public repository names unless its owner names others, as `dev.knotless.describeServer` answers; one only, as each runner a repository names runs all its checks (`--default-runner names one runner`); empty, none; anything else not a DID is refused (`--default-runner takes a DID, not <value>`) | | ||
harness/star-fixture.mjsnew+73−0
| @@ -0,0 +1,73 @@ | ||
// Makes the repository knotless-serve's archive tests read: a few dozen | ||
// records, the MST made by @atproto/repo, as atproto's reference makes it, | ||
// written as a CAR and as STAR-lite version 0 (microcosm's, as hubble | ||
// serves it), with the DID and the key that signed its commit. | ||
// | ||
// node harness/star-fixture.mjs crates/knotless-serve/fixtures | ||
// | ||
// STAR-lite: `*l\0`, the MST root's CID (36 bytes), the commit without its | ||
// `data` (a varint length, then DRISL), then each record in key order as a | ||
// varint key length, the key, a varint record length and the record. | ||
// https://tangled.org/microcosm.blue/star/tree/main/v0-star/star-lite | ||
| ||
import { writeFileSync } from 'node:fs' | ||
import { join } from 'node:path' | ||
import { encode } from '@atproto/lex-cbor' | ||
import { Secp256k1Keypair } from '@atproto/crypto' | ||
import { MemoryBlockstore, Repo, WriteOpAction, blocksToCarFile } from '@atproto/repo' | ||
| ||
const out = process.argv[2] | ||
if (!out) throw new Error('usage: node harness/star-fixture.mjs <dir>') | ||
| ||
// A fixed key, so the fixture's DID stays the same each time it is made. | ||
const keypair = await Secp256k1Keypair.import( | ||
'9085d2bef69286a6cbb51623c8fa258629945cd55ca705cc4e66700396894e0c', | ||
) | ||
const did = 'did:plc:archivefixture00000000000' | ||
const writes = [] | ||
for (let i = 0; i < 30; i++) { | ||
const rkey = `3l${String(i).padStart(11, 'a')}` | ||
writes.push({ | ||
action: WriteOpAction.Create, | ||
collection: 'dev.knotless.repo', | ||
rkey, | ||
record: { $type: 'dev.knotless.repo', name: `r${i}`, key: `k${i}`, createdAt: '2026-11-01T00:00:00.000Z' }, | ||
}) | ||
writes.push({ | ||
action: WriteOpAction.Create, | ||
collection: 'app.bsky.feed.post', | ||
rkey, | ||
record: { $type: 'app.bsky.feed.post', text: `post ${i}`, createdAt: '2026-11-01T00:00:00.000Z' }, | ||
}) | ||
} | ||
const storage = new MemoryBlockstore() | ||
const repo = await Repo.create(storage, did, keypair, writes) | ||
| ||
const car = await blocksToCarFile(repo.cid, storage.blocks) | ||
| ||
const varint = (n) => { | ||
const bytes = [] | ||
while (n >= 0x80) { | ||
bytes.push((n & 0x7f) | 0x80) | ||
n >>>= 7 | ||
} | ||
bytes.push(n) | ||
return Buffer.from(bytes) | ||
} | ||
const { data, ...partial } = repo.commit | ||
const commit = Buffer.from(encode(partial)) | ||
const parts = [Buffer.from([0x2a, 0x6c, 0x00]), Buffer.from(data.bytes), varint(commit.length), commit] | ||
for (const leaf of await repo.data.list()) { | ||
const record = Buffer.from(await storage.getBytes(leaf.value)) | ||
const key = Buffer.from(leaf.key, 'ascii') | ||
parts.push(varint(key.length), key, varint(record.length), record) | ||
} | ||
const star = Buffer.concat(parts) | ||
| ||
writeFileSync(join(out, 'repo.car'), car) | ||
writeFileSync(join(out, 'repo.star'), star) | ||
writeFileSync( | ||
join(out, 'repo.json'), | ||
`${JSON.stringify({ did, key: keypair.did(), rev: repo.commit.rev, records: writes.length }, null, 2)}\n`, | ||
) | ||
console.log(`${writes.length} records, CAR ${car.length} bytes, STAR-lite ${star.length} bytes`) | ||