[db] write v10 bodies into the compressed heads keyspace
v9 created records with compression disabled at every level and fjall persists keyspace options on disk, so bodies written there stay uncompressed no matter what the schema says at open (measured on the prod db copy: 71.3MB stored vs 16MB zstd:3 standalone on the same bodies). the schema registers the store under a new name, heads, with zstd at every level, and v10 stages rewritten bodies straight into it, retiring both legacy stores in its finalize. chunked passes now resolve scan keyspaces through an on-disk fallback for stores the registry has dropped, pass the scanned keyspace to visit so in-place rewrites keep writing the store they read, and treat a never-existing source as a no-op when the pass is optional_on_absent. verified on a copy of the prod database: 136,311 entries migrated in ~500ms, heads compacts to 13.1MB (5.4x smaller), record resolution unchanged.
94 unmodified lines9596979899991001011021037 unmodified lines111112113114115115116117118118119120121122178 unmodified linesExpand all94 unmodified lines- Avoid unnecessary comments if the code is self-documenting.
### Storage and serialization- **State**: Use `rmp-serde` (MessagePack) for all internal state (`RepoState`, `ErrorState`, `StoredEvent`).- **Blocks**: Record bodies are stored inline in `records`; superseded bodies move to `history`. v10 materializes unresolved legacy event bodies into `event_bodies`, then deletes the old `blocks` keyspace.- **Blocks**: Record bodies are stored inline in `heads`; superseded bodies move to `history`. v10 materializes unresolved legacy event bodies into `event_bodies`, moves bodies into the compressed `heads` keyspace, and deletes the old `blocks` and `records` keyspaces.- **Cursors**: Store cursors as big-endian bytes (`u64`/`i64`).- **Compression**: Configurable via `HYDRANT_DATA_COMPRESSION` (`lz4`, `zstd`, `none`). Per-keyspace zstd dictionaries can be trained via `POST /db/train` and are stored as `dict_{keyspace}.bin` in the database directory.- **Keyspaces**: Use the `keys.rs` module to maintain consistent composite key formats.- **Schema evolution**: Treat versioned wire types in `src/types.rs` as frozen snapshots once shipped. If a stored type changes shape, add a new versioned type, export the newest version for live code, and add a forward migration that explicitly deserializes the previous version and writes the new one. Do not modify older migration input/output types in place. For example, if `RepoState` changes after `v7`, add `v8::RepoState` and a `v7 -> v8` migration instead of mutating `v7`.7 unmodified lines
Hydrant uses multiple `fjall` keyspaces:- `repos`: Maps `{DID}` -> `RepoState` (MessagePack).- `repo_metadata`: Maps `rm|{DID}` -> `RepoMetadata` (MessagePack).- `records`: Maps `{DID} 00 {COL} 2f {RKey}` (MST order, trimmed DID) -> record body (raw DAG-CBOR). Operator-redacted heads retain only their raw CID bytes; links-only mode stores CIDs for every head. Values were CIDs before v10.- `heads`: Maps `{DID} 00 {COL} 2f {RKey}` (MST order, trimmed DID) -> record body (raw DAG-CBOR), zstd-compressed at every level. Operator-redacted heads retain only their raw CID bytes; links-only mode stores CIDs for every head. Replaces v9's `records` keyspace in the v10 migration (v9 persisted "never compress" into the old keyspace's on-disk config); values were CIDs before v10.- `history`: Maps `{record key} 00 {death_rev}` (rev = 8-byte BE TID) -> superseded record body; raw CID bytes = operator-redacted body; empty value = delete tombstone ("deleted at rev", vs never existed). Retention is bounded by `HYDRANT_HISTORY_TTL` via a key-only compaction filter (`db::compaction::HistoryRetentionFilterFactory`); unset = permanent. Ephemeral mode stores no history.- `redactions`: Maps `{record key} 00 {CID bytes}` -> empty for operator-erased versions. Markers are permanent and suppress the same record+CID on replay even after an intervening version or history compaction.- `event_bodies` (indexer stream): Finite compatibility archive, `{record key} 00 {CID bytes}` -> raw DAG-CBOR. v10 uses it only for legacy events that cannot resolve through `records` or `history`; new writes never add entries.- `event_bodies` (indexer stream): Finite compatibility archive, `{record key} 00 {CID bytes}` -> raw DAG-CBOR. v10 uses it only for legacy events that cannot resolve through `heads` or `history`; new writes never add entries.- `events`: Maps `{ID}` (u64 BE) -> `StoredEvent` (MessagePack). This is the source for the JSON stream API. Ephemeral mode stores create/update bodies inline as `StoredData::Block`, so event TTL removes identity and body bytes together without creating record heads or history.- `cursors`: Maps per-relay cursor keys -> `Value` (u64/i64 BE Bytes). Keys: `firehose_cursor|{relay}`, `crawler_cursor|{relay}`, `by_collection_cursor|{url}|{collection}`.- `pending`: Queue of `{ID}` (u64 BE) -> `Empty` (Backfill queue).- `resync`: Maps `{DID}` -> `ResyncState` (MessagePack) for retry logic/tombstones.178 unmodified linesExpand all22 unmodified lines23242526272827282930313213 unmodified lines4647484950505152535448 unmodified lines22 unmodified lines if (wait-for-backfill $url) { print "backfill complete, testing debug endpoints"
# 1. Test /debug/iter to find a key print "testing /debug/iter on records partition" let records = http get $"($debug_url)/debug/iter?partition=records&limit=1" print "testing /debug/iter on heads partition" let records = http get $"($debug_url)/debug/iter?partition=heads&limit=1" if ($records.items | is-empty) { print "FAILED: /debug/iter returned empty items" exit 113 unmodified lines
# 2. Test /debug/get with that key (sent as string) print "testing /debug/get" let encoded_key = ($key_str | url encode) let get_res = http get $"($debug_url)/debug/get?partition=records&key=($encoded_key)" let get_res = http get $"($debug_url)/debug/get?partition=heads&key=($encoded_key)" if $get_res.value != $value_cid { print $"FAILED: /debug/get returned different value. expected: ($value_cid), got: ($get_res.value)" exit 148 unmodified lines4 unmodified lines567899101112131415976 unmodified linesExpand all4 unmodified lines`hydrant-u93`, `hydrant-w8q`, `hydrant-iis`
## Decision update
The permanent layout below landed: current bodies live inline in `records`,The permanent layout below landed: current bodies live inline in `heads`(the v10 migration builds that keyspace with zstd at every level — v9 hadpersisted "never compress" into the old `records` keyspace's on-disk config),superseded bodies move to `history`, and `HYDRANT_HISTORY_TTL` optionally boundsthat history. The queryable-ephemeral extension did not land. Ephemeral remainsthe shipped separate architecture: it stores create/update bodies only insidemonotonic `events`, exposes no record-body XRPCs, and event TTL removes metadata976 unmodified linesExpand all50 unmodified lines51525354555556575859236 unmodified linesExpand all50 unmodified lines#[cfg(feature = "indexer")]#[derive(Clone)]pub struct IndexerDb { /// maps `{DID} 00 {COL} 2f {RKey}` -> DAG-CBOR body or redacted CID marker pub(super) records: Ks<schema::Records>, pub(super) records: Ks<schema::Heads>, /// superseded bodies/markers: `{record key} 00 {death_rev}` -> body/CID/empty pub(super) history: Ks<schema::History>, /// operator-redacted versions: `{record key} 00 {cid}` -> empty pub(super) redactions: Ks<schema::Redactions>,236 unmodified linesExpand all94 unmodified lines95969798999910010110210317 unmodified lines1211221231241251251261271281291291301311321332 unmodified lines94 unmodified lines crawler => schema::Crawler, #[cfg(feature = "backlinks")] backlinks => schema::Backlinks, #[cfg(feature = "indexer")] indexer.records => schema::Records, indexer.records => schema::Heads, #[cfg(feature = "indexer")] indexer.history => schema::History, #[cfg(feature = "indexer")] indexer.redactions => schema::Redactions,17 unmodified linesmod tests { #[test] fn record_debug_key_roundtrips() { let parsed = super::debug_parse_key( "records", "heads", "web:guestbook.gaze.systems|app.bsky.actor.profile|self", ); assert!(parsed.is_some(), "parse failed"); let rendered = super::debug_render_key("records", &parsed.unwrap()); let rendered = super::debug_render_key("heads", &parsed.unwrap()); assert_eq!( rendered, "web:guestbook.gaze.systems|app.bsky.actor.profile|self" );2 unmodified lines321 unmodified linesExpand all3223233243253263263273283293303293303313323333345 unmodified lines34034134234334434534434534634734834935035135235334734834935035435535635735835936036173 unmodified lines43543643743843243944044144244329 unmodified lines47347447547647047747847948048131 unmodified lines513514515516510517518519520521237 unmodified linesExpand all321 unmodified linesExpand all}
/// maps `{DID} 00 {COL} 2f {RKey}` -> DAG-CBOR body or redacted CID marker.#[cfg(feature = "indexer")]pub struct Records;pub struct Heads;
#[cfg(feature = "indexer")]impl Schema for Records { const NAME: &'static str = "records";impl Schema for Heads { const NAME: &'static str = "heads";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // point reads might miss when using getRecord5 unmodified lines // did 00 col 2f rkey -> record body (dag-cbor). bodies compress well // grouped by repo: author/time locality beats the old col|cid // content addressing (-56% measured on a live corpus, zstd:3@64k) .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(64), kb(128)])) // leave the first level uncompressed; head bodies of fresh records // are hot for stream inflation // compress from the first level: bulk loads (migrations) write in // sorted order, so their tables never overlap and compaction // trivial-moves them without ever re-compressing — a None first // level sticks to the whole tree (measured: 71MB stored vs 16MB // zstd:3 standalone on the same bodies). heads stay hot in the // memtable regardless. v10 migrates pre-v10 databases into this // keyspace because v9's `records` persisted "never compress" into // fjall's on-disk config, which no open-time schema change can // undo. .data_block_compression_policy(CompressionPolicy::new([ CompressionType::None, (cx.compression)("records", 3), (cx.compression)("records", 3), (cx.compression)("records", 5), (cx.compression)("heads", 3), (cx.compression)("heads", 3), (cx.compression)("heads", 3), (cx.compression)("heads", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])) }
73 unmodified lines return String::from_utf8_lossy(key).into_owned(); } let rev_bytes: [u8; 8] = key[key.len() - 8..].try_into().expect("length checked"); let rev = crate::db::types::DbTid::new_from_bytes(rev_bytes); format!("{}@{rev}", Records::render_debug_key(&key[..key.len() - 9])) format!("{}@{rev}", Heads::render_debug_key(&key[..key.len() - 9])) }}
/// durable operator-redaction identities: `{record key} 00 {cid}` -> empty.29 unmodified lines fn render_debug_key(key: &[u8]) -> String { let Some((record_key, cid)) = super::keys::indexer::split_record_cid_key(key) else { return String::from_utf8_lossy(key).into_owned(); }; format!("{}#{cid}", Records::render_debug_key(record_key)) format!("{}#{cid}", Heads::render_debug_key(record_key)) }}
/// record-scoped bodies retained solely for legacy stream events:31 unmodified lines fn render_debug_key(key: &[u8]) -> String { let Some((record_key, cid)) = super::keys::indexer::split_record_cid_key(key) else { return String::from_utf8_lossy(key).into_owned(); }; format!("{}#{cid}", Records::render_debug_key(record_key)) format!("{}#{cid}", Heads::render_debug_key(record_key)) }}
/// backfill queue of `{ID}` (u64 BE) -> empty237 unmodified linesExpand all36 unmodified lines37383940414243444546474849505152535455565758596061626364656667689 unmodified lines7879808182838485868758888990919263 unmodified lines15615715815913013116016116216316416516616716816917017117217317417517617717817950 unmodified lines230231232233197234235236237238117 unmodified linesExpand all35635735835932336036136236336416 unmodified lines381382383384348385386387388389390391392393394395396397398399400401402403404405406407408409410138 unmodified linesExpand all5495505515525535545555565575585595605615625635045055645655665675685694 unmodified lines574575576577578579580581582114 unmodified linesExpand all6976986997007017027037047052 unmodified lines708709710711650712713714715716208 unmodified linesExpand all36 unmodified lines .keyspace(LEGACY_BLOCKS_KEYSPACE, Default::default) .into_diagnostic()}
/// the pre-v12 record store. v12's schema drops `records` from the registry,/// but its on-disk tables persist until the `recompress_record_heads`/// migration copies them into `heads` and deletes the keyspace.#[cfg(feature = "indexer")]const LEGACY_RECORDS_KEYSPACE: &str = "records";
#[cfg(all(test, feature = "indexer"))]fn legacy_records_for_test(db: &Db) -> Result<Keyspace> { db.inner .keyspace(LEGACY_RECORDS_KEYSPACE, Default::default) .into_diagnostic()}
#[cfg(feature = "indexer")]fn legacy_records(db: &Db) -> Result<Option<Keyspace>> { if !db.inner.keyspace_exists(LEGACY_RECORDS_KEYSPACE) { return Ok(None); } db.inner .keyspace(LEGACY_RECORDS_KEYSPACE, Default::default) .into_diagnostic() .map(Some)}
mod v1;mod v10;mod v2;mod v3;9 unmodified linestype AtomicFn = fn(&Db, &mut OwnedWriteBatch) -> Result<bool>;
/// transform one scanned entry. staged writes may target any keyspace./// returns the number of output bytes staged into the write batch for this entry./// visit one scanned entry, staging its rewrite into the chunk batch, and/// return the staged byte count for the chunk budget.////// the third argument is the keyspace being scanned: in-place passes stage/// into it, so a pass keeps writing the same store it reads even when that/// store is a legacy keyspace the registry no longer knows.type VisitFn = fn(&Db, &mut OwnedWriteBatch, &[u8], &[u8]) -> Result<usize>;type VisitFn = fn(&Db, &mut OwnedWriteBatch, &Keyspace, &[u8], &[u8]) -> Result<usize>;
/// destructive or otherwise non-batched work that runs only after every/// chunked pass is durable. returning false means this build does not contain/// the keyspaces needed by the migration.63 unmodified lines /// stable cursor identity. THIS SHOULD ALWAYS BE STABLE. DO NOT CHANGE. name: &'static str, /// on-disk name of the keyspace to scan. /// /// a pass whose keyspace is not compiled into this build is skipped, so a /// migration can name mode-specific keyspaces unconditionally. /// the registry resolves names from this build first; a name it no longer /// knows falls back to opening an existing on-disk keyspace, so a pass can /// still finish migrating data out of a store a later schema dropped. /// /// a pass whose keyspace exists nowhere is skipped, so a migration can /// name mode-specific keyspaces unconditionally — unless /// `optional_on_absent` is set, in which case absence means the database /// never had the source layout and the pass completes as a no-op. /// /// a pass scanning `counts` also sees its own resume cursor, which sorts /// between the `k|` and `r|` prefixes; `visit` must ignore keys under /// [`MIGRATION_CURSOR_PREFIX`]. scan: &'static str, visit: VisitFn, budget: ChunkBudget, optional_on_absent: bool,}
/// ordered list of schema migrations.///50 unmodified lines ( "inline_record_bodies", Migration::Chunked { passes: v10::PASSES, finalize: Some(v10::retire_blocks), finalize: Some(v10::retire_legacy_stores), }, ),];
117 unmodified linesExpand all exhausted = false; break; } let (key, value) = guard.into_inner().into_diagnostic()?; let staged_bytes = (pass.visit)(db, &mut batch, &key, &value) let staged_bytes = (pass.visit)(db, &mut batch, ks, &key, &value) .wrap_err_with(|| format!("migration pass {} failed on an entry", pass.name))?; bytes += staged_bytes; last = Some(key); seen += 1;16 unmodified lines}
/// run one pass to completion, resuming from its stored cursor if present.fn run_pass(db: &Db, version: usize, pass: &Pass) -> Result<bool> { let Some(ks) = db.keyspace_by_name(pass.scan) else { let ks = match db.keyspace_by_name(pass.scan) { Some(ks) => Some(ks), // a keyspace dropped from the registry by a later schema may still // exist on disk; open it so the pass can finish migrating its data None if db.inner.keyspace_exists(pass.scan) => Some( db.inner .keyspace(pass.scan, Default::default) .into_diagnostic()?, ), None => None, }; let Some(ks) = ks else { if pass.optional_on_absent { // the source layout never existed on this database, so there is // genuinely nothing to migrate tracing::debug!( "db: migration pass {} found no {} keyspace, nothing to migrate", pass.name, pass.scan ); return Ok(true); } // keyspace is not compiled into this build, so there is nothing to migrate tracing::debug!( "db: migration pass {} skipped, keyspace {} not in this build", pass.name,138 unmodified linesExpand all /// the usual shape for a key rewrite and means output is never rescanned. fn rewrite_prefix( db: &Db, batch: &mut OwnedWriteBatch, scanned: &Keyspace, key: &[u8], value: &[u8], ) -> Result<usize> { let Some(suffix) = key.strip_prefix(TEST_PREFIX) else { return Ok(0); }; let mut new_key = MIGRATED_PREFIX.to_vec(); new_key.extend_from_slice(suffix); let _ = db; let staged_bytes = new_key.len() + value.len(); batch.insert(&db.cursors, new_key, value); batch.remove(&db.cursors, key); batch.insert(scanned, new_key, value); batch.remove(scanned, key); Ok(staged_bytes) }
fn test_pass(entries: usize) -> Pass {4 unmodified lines budget: ChunkBudget { entries, bytes: usize::MAX, }, optional_on_absent: false, } }
fn open_db() -> Result<(tempfile::TempDir, Db)> {114 unmodified linesExpand all /// fails partway through a chunk, standing in for a crash mid-chunk. fn fail_partway( db: &Db, batch: &mut OwnedWriteBatch, scanned: &Keyspace, key: &[u8], value: &[u8], ) -> Result<usize> { let Some(suffix) = key.strip_prefix(TEST_PREFIX) else {2 unmodified lines let n = u64::from_be_bytes(suffix.try_into().into_diagnostic()?); if n == 4 { miette::bail!("simulated failure at entry 4"); } rewrite_prefix(db, batch, key, value) rewrite_prefix(db, batch, scanned, key, value) } #[test] fn a_chunk_that_fails_partway_commits_nothing() -> Result<()> { let (_tmp, db) = open_db()?;208 unmodified linesExpand all11233456456789101112131491011121315161718192021222324 unmodified lines4849505152535455562 unmodified lines5960616263646566676812 unmodified lines8182838476858687888990915 unmodified lines979899100919210110210310410510612 unmodified lines11912012112211311412312412512612712812913013113213311813413513613713813914014114214314414514614714814915015134 unmodified lines18618718818916119019119219319419519619719819916320020120220320420536 unmodified lines24224324424520824624724824965 unmodified lines3153163173183193203213223232862872883243253263273283293303 unmodified lines33433533633730033833934034134259 unmodified lines40240340440540640740840941041150 unmodified lines46246346446546646742746846947047147224 unmodified lines4974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575582 unmodified lines5615625635645655665675685695 unmodified lines5755765775784835795805815825838 unmodified lines5925935945955965975985025996006016026033 unmodified lines//! v10: inline record bodies and retire the legacy blocks CAS.//! v10: inline record bodies into the compressed `heads` keyspace, and retire//! the legacy `records` store and `blocks` CAS.//!//! legacy `{DID}|{COL}|{tag}{rkey}` -> cid entries become//! `{DID} 00 {COL} 2f {rkey text}` -> body, resolved from the blocks CAS.//! new keys sort before legacy keys within a repo (NUL < `|` after the//! trimmed DID), so the resuming chunked scan only ever meets legacy keys.//! legacy `{DID}|{COL}|{tag}{rkey}` -> cid entries in `records` become//! `{DID} 00 {COL} 2f {rkey text}` -> body in `heads`, resolved from the//! blocks CAS. the schema drops `records` from the registry for this: v9//! persisted "never compress" into its on-disk config (fair when values were//! cids), and fjall reads keyspace options from disk at open, so bodies could//! only ever be compressed in a fresh keyspace. the pass scans `records`//! through the on-disk fallback and stages into a different keyspace than it//! scans, so chunks never re-observe their own output and re-inserting the//! same key/value is a fixpoint.//!//! the same bounded migration archives otherwise-unresolved legacy permanent//! pointer bodies before deleting `blocks`. ephemeral inline events stay//! inline so their TTL still removes body bytes atomically. no ingestion can//! run during `Db::open`, so every pointer visible to this migration is//! necessarily legacy; an event watermark would only model versions that//! never shipped.//! pointer bodies before deleting `blocks` and `records`. ephemeral inline//! events stay inline so their TTL still removes body bytes atomically. no//! ingestion can run during `Db::open`, so every pointer visible to this//! migration is necessarily legacy; an event watermark would only model//! versions that never shipped.
mod events;
#[cfg(feature = "indexer")]24 unmodified lines name: "finish_v9_filter_layout", scan: "filter", visit: rewrite_filter_key, budget: ChunkBudget::DEFAULT, optional_on_absent: false, }, #[cfg(feature = "indexer")] Pass { name: "records",2 unmodified lines budget: ChunkBudget { entries: 100_000, bytes: DEFAULT_RECORD_CHUNK_BYTES, }, // v12 databases register `heads` instead and never had `records` optional_on_absent: true, }, #[cfg(feature = "indexer_stream")] events::POINTER_EVENT_BODIES,];12 unmodified lines}
#[cfg(feature = "indexer")]fn rewrite_filter_key( db: &Db, _db: &Db, batch: &mut OwnedWriteBatch, scanned: &fjall::Keyspace, key: &[u8], value: &[u8],) -> Result<usize> { let exclude_prefix = [crate::db::filter::EXCLUDE_PREFIX, crate::db::keys::SEP];5 unmodified lines let trimmed = TrimmedDid::from(&did); let mut new_key = exclude_prefix.to_vec(); trimmed.write_to_vec(&mut new_key); let staged = new_key.len(); batch.insert(&db.filter, new_key, []); batch.remove(&db.filter, key); batch.insert(scanned, new_key, []); batch.remove(scanned, key); return Ok(staged); }
if key.starts_with(b"pds|status|") || key.starts_with(b"pds|tier|") {12 unmodified lines let mut new_key = Vec::with_capacity(current_prefix.len() + host.len()); new_key.extend_from_slice(current_prefix); new_key.extend_from_slice(host); let staged = new_key.len() + value.len(); batch.insert(&db.filter, new_key, value); batch.remove(&db.filter, key); batch.insert(scanned, new_key, value); batch.remove(scanned, key); Ok(staged)}
#[cfg(feature = "indexer")]/// delete the legacy `records` keyspace only after every copy chunk and its/// resume cursor are durable. chunk commits use the data WAL while keyspace/// deletion updates fjall's metadata independently, so force the copy durable/// before making the old store unreachable. fjall removes the on-disk/// directory on the next open, when recovery finds it unreferenced.pub(super) fn retire_blocks(db: &crate::db::Db) -> miette::Result<bool> {pub(super) fn retire_legacy_stores(db: &crate::db::Db) -> miette::Result<bool> { if !events::retire_blocks(db)? { return Ok(false); } if let Some(records) = super::legacy_records(db)? { db.persist()?; db.inner.delete_keyspace(records).into_diagnostic()?; db.persist()?; } Ok(true)}
#[cfg(not(feature = "indexer"))]pub(super) fn retire_legacy_stores(db: &crate::db::Db) -> miette::Result<bool> { events::retire_blocks(db)}
#[cfg(feature = "indexer")]34 unmodified lines Ok((did_len, collection, rkey))}
#[cfg(feature = "indexer")]fn rewrite_record(db: &Db, batch: &mut OwnedWriteBatch, key: &[u8], value: &[u8]) -> Result<usize> {fn rewrite_record( db: &Db, batch: &mut OwnedWriteBatch, _scanned: &fjall::Keyspace, key: &[u8], value: &[u8],) -> Result<usize> { // already-current key (defense against out-of-order surprises): the body // is final, copy it verbatim into heads if is_current_key(key) { return Ok(0); batch.insert(&db.indexer.records, key, value); return Ok(key.len() + value.len()); } let (did_len, collection, rkey) = parse_legacy_key(key)?;
// keep the trimmed-DID bytes, swap the separators, re-encode the rkey as text36 unmodified lines };
let staged_bytes = new_value.len(); batch.insert(&db.indexer.records, new_key, new_value); batch.remove(&db.indexer.records, key); Ok(staged_bytes)}
#[cfg(all(test, feature = "indexer"))]65 unmodified lines { let db = Db::open(&cfg)?; let mut batch = db.inner.batch(); let blocks = super::super::legacy_blocks_for_test(&db)?; let records = super::super::legacy_records_for_test(&db)?;
let post_legacy = legacy_record_key(&did(), "app.bsky.feed.post", &tid_rkey); let profile_legacy = legacy_record_key(&did(), "app.bsky.actor.profile", &str_rkey); let orphan_legacy = legacy_record_key(&did(), "app.bsky.feed.like", &tid_rkey); batch.insert(&db.indexer.records, &post_legacy, &cid1); batch.insert(&db.indexer.records, &profile_legacy, &cid2); batch.insert(&db.indexer.records, &orphan_legacy, &cid3); batch.insert(&records, &post_legacy, &cid1); batch.insert(&records, &profile_legacy, &cid2); batch.insert(&records, &orphan_legacy, &cid3);
// bodies live in the blocks cas batch.insert(&blocks, keys::block_key("app.bsky.feed.post", &cid1), body1); batch.insert(3 unmodified lines );
// an already-current record: the pass must leave it alone let current_key = keys::record_key(&did(), "app.bsky.feed.repost", &str_rkey); batch.insert(&db.indexer.records, ¤t_key, b"repost body"); batch.insert(&records, ¤t_key, b"repost body");
// a body-less legacy event remains valid through the event passes #[cfg(feature = "indexer_stream")] {59 unmodified lines );
// v10 retires the frozen cas after event compatibility is established assert!(!db.inner.keyspace_exists("blocks")); // and the legacy records store, whose contents moved to heads assert!(!db.inner.keyspace_exists("records"));
Ok(()) }
50 unmodified lines
{ let db = Db::open(&cfg)?; let blocks = super::super::legacy_blocks_for_test(&db)?; let records = super::super::legacy_records_for_test(&db)?; let mut batch = db.inner.batch(); batch.insert(&db.indexer.records, &legacy_key, &cid); batch.insert(&records, &legacy_key, &cid); batch.insert( &blocks, keys::block_key("app.bsky.actor.profile", &cid), wrong,24 unmodified lines Ok(()) }
#[test] fn v10_heads_are_actually_compressed() -> Result<()> { // v9's records keyspace persisted "never compress" into its on-disk // config; the migrated bodies must land in a store that compresses let tmp = tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(), ..Default::default() };
// highly compressible bodies, larger than one data block let body = vec![7u8; 4096]; let count = 2000u32; let raw: u64; { let db = Db::open(&cfg)?; let records = super::super::legacy_records_for_test(&db)?; let mut batch = db.inner.batch(); for n in 0..count { let key = keys::record_key( &Did::new(&format!("did:plc:{n:016x}")).unwrap(), "app.bsky.feed.post", &DbRkey::new(&format!("r{n}")), ); batch.insert(&records, key, &body); } batch.commit().into_diagnostic()?; raw = (0..count) .map(|n| { (keys::record_key( &Did::new(&format!("did:plc:{n:016x}")).unwrap(), "app.bsky.feed.post", &DbRkey::new(&format!("r{n}")), ) .len() + body.len()) as u64 }) .sum(); crate::db::migration::rewind_version_for_test(&db, 9)?; db.persist()?; }
let db = Db::open(&cfg)?; db.indexer.records.major_compact().into_diagnostic()?; db.persist()?;
let stored = db.indexer.records.disk_space(); assert!( stored < raw / 2, "heads should compress repeated bodies: stored {stored} vs raw {raw}" ); Ok(()) }
#[test] fn v10_byte_budget_chunking_limits_output_bytes() -> Result<()> { let tmp = tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(),2 unmodified lines
let db = Db::open(&cfg)?; let mut batch = db.inner.batch(); let blocks = super::super::legacy_blocks_for_test(&db)?; let records = super::super::legacy_records_for_test(&db)?;
let large_body = vec![7u8; 10_000]; // 10 KB body // tid alphabet is base32-sortable (no 0/1/8/9 digits) const TID_CHARS: &[u8] = b"234567abcdefghij";5 unmodified lines let cid = jacquard_repo::mst::util::compute_cid(&large_body) .into_diagnostic()? .to_bytes(); let legacy_key = legacy_record_key(&did(), "app.bsky.feed.post", &tid_rkey); batch.insert(&db.indexer.records, &legacy_key, &cid); batch.insert(&records, &legacy_key, &cid); batch.insert( &blocks, keys::block_key("app.bsky.feed.post", &cid), &large_body,8 unmodified lines budget: super::ChunkBudget { entries: usize::MAX, bytes: 25_000, // 25 KB budget }, optional_on_absent: true, }; let cursor_key = super::super::pass_cursor_key(9, pass.name); let ks = db.keyspace_by_name("records").expect("records keyspace"); let ks = super::super::legacy_records_for_test(&db)?;
let outcome = super::super::run_chunk(&db, &cursor_key, &ks, &pass)?; assert_eq!(outcome.seen, 3); assert!(!outcome.exhausted);3 unmodified lines25 unmodified lines26272829303132333438 unmodified lines737475767778798081509 unmodified linesExpand all591592593594595596597598599241 unmodified linesExpand all25 unmodified lines name: "pointer_event_bodies", scan: "events", visit: materialize_pointer_event, budget: ChunkBudget::DEFAULT, optional_on_absent: false,};
#[cfg(feature = "indexer_stream")]fn body_matches(value: &[u8], expected: &jacquard_common::types::cid::IpldCid) -> bool {38 unmodified lines#[cfg(feature = "indexer_stream")]fn materialize_pointer_event( db: &Db, batch: &mut OwnedWriteBatch, _scanned: &fjall::Keyspace, _key: &[u8], value: &[u8],) -> Result<usize> { let event = decode_event(value)?;509 unmodified linesExpand all budget: ChunkBudget { entries: 1, bytes: usize::MAX, }, optional_on_absent: false, }; let cursor = super::super::super::pass_cursor_key(9, pass.name); let events = db.keyspace_by_name("events").expect("events keyspace"); let outcome = super::super::super::run_chunk(&db, &cursor, &events, &pass)?;241 unmodified linesExpand all