+23 -23

src/backfill/mod.rs

··· 9 use jacquard::types::did::Did; 10 use jacquard::{CowStr, IntoStatic, prelude::*}; 11 use jacquard_common::xrpc::XrpcError; 12 - use jacquard_repo::MemoryBlockStore; 13 use jacquard_repo::mst::Mst; 0 14 use miette::{IntoDiagnostic, Result}; 15 use smol_str::{SmolStr, ToSmolStr, format_smolstr}; 16 use std::collections::HashMap; ··· 305 emit_identity(&state.status); 306 307 trace!( 308 - "fetched {} bytes for {} in {:?}", 309 car_bytes.body.len(), 310 - did, 311 start.elapsed() 312 ); 313 ··· 316 let parsed = jacquard_repo::car::reader::parse_car_bytes(&car_bytes.body) 317 .await 318 .into_diagnostic()?; 319 - trace!("parsed car for {} in {:?}", did, start.elapsed()); 320 321 let start = Instant::now(); 322 - let store = Arc::new(MemoryBlockStore::new()); 323 - for (_cid, bytes) in &parsed.blocks { 324 - jacquard_repo::BlockStore::put(store.as_ref(), bytes) 325 - .await 326 - .into_diagnostic()?; 327 - } 328 trace!( 329 - "stored {} blocks in memory for {} in {:?}", 330 - parsed.blocks.len(), 331 - did, 332 start.elapsed() 333 ); 334 335 // 4. parse root commit to get mst root 336 - let root_bytes = parsed 337 - .blocks 338 .get(&parsed.root) 0 0 339 .ok_or_else(|| miette::miette!("root block missing from CAR"))?; 340 341 - let commit = jacquard_repo::commit::Commit::from_cbor(root_bytes).into_diagnostic()?; 342 - debug!("backfilling repo at revision {}", commit.rev); 0 0 0 0 343 344 // 4.5. verify commit signature 345 if verify_signatures { 346 let pubkey = app_state.resolver.resolve_signing_key(did).await?; 347 - commit 348 .verify(&pubkey) 349 .map_err(|e| miette::miette!("signature verification failed for {did}: {e}"))?; 350 trace!("signature verified for {did}"); ··· 352 353 // 5. walk mst 354 let start = Instant::now(); 355 - let mst: Mst<MemoryBlockStore> = Mst::load(store, commit.data, None); 356 let leaves = mst.leaves().await.into_diagnostic()?; 357 trace!("walked mst for {} in {:?}", did, start.elapsed()); 358 ··· 361 let (_state, added_records, added_blocks, count) = { 362 let app_state = app_state.clone(); 363 let did = did.clone(); 364 - let rev = commit.rev; 365 - let storage = mst.storage().clone(); 366 367 tokio::task::spawn_blocking(move || { 368 let mut count = 0; ··· 370 let mut added_blocks = 0; 371 let mut collection_counts: HashMap<SmolStr, u64> = HashMap::new(); 372 let mut batch = app_state.db.inner.batch(); 0 373 374 // pre-load existing record CIDs for this DID to detect duplicates/updates 375 let prefix = keys::record_prefix(&did); ··· 385 386 for (key, cid) in leaves { 387 let val_bytes = tokio::runtime::Handle::current() 388 - .block_on(jacquard_repo::BlockStore::get(storage.as_ref(), &cid)) 389 .into_diagnostic()?; 390 391 if let Some(val) = val_bytes { ··· 398 let (action, is_new) = if let Some(existing_cid) = existing_cids.get(&path) 399 { 400 if existing_cid == cid.as_str() { 0 401 continue; // skip unchanged record 402 } 403 ("update", false) 404 } else { 405 ("create", true) 406 }; 0 407 408 let db_key = keys::record_key(&did, &collection, rkey); 409 ··· 440 // 6. update status to synced 441 state.status = RepoStatus::Synced; 442 state.rev = Some(rev); 443 - state.data = Some(Cid::ipld(parsed.root)); 444 state.last_updated_at = chrono::Utc::now().timestamp(); 445 446 batch.insert(

··· 9 use jacquard::types::did::Did; 10 use jacquard::{CowStr, IntoStatic, prelude::*}; 11 use jacquard_common::xrpc::XrpcError; 0 12 use jacquard_repo::mst::Mst; 13 + use jacquard_repo::{BlockStore, MemoryBlockStore}; 14 use miette::{IntoDiagnostic, Result}; 15 use smol_str::{SmolStr, ToSmolStr, format_smolstr}; 16 use std::collections::HashMap; ··· 305 emit_identity(&state.status); 306 307 trace!( 308 + "fetched {} bytes for {did} in {:?}", 309 car_bytes.body.len(), 0 310 start.elapsed() 311 ); 312 ··· 315 let parsed = jacquard_repo::car::reader::parse_car_bytes(&car_bytes.body) 316 .await 317 .into_diagnostic()?; 318 + trace!("parsed car for {did} in {:?}", start.elapsed()); 319 320 let start = Instant::now(); 321 + let store = Arc::new(MemoryBlockStore::new_from_blocks(parsed.blocks)); 0 0 0 0 0 322 trace!( 323 + "stored {} blocks in memory for {did} in {:?}", 324 + store.len(), 0 325 start.elapsed() 326 ); 327 328 // 4. parse root commit to get mst root 329 + let root_bytes = store 0 330 .get(&parsed.root) 331 + .await 332 + .into_diagnostic()? 333 .ok_or_else(|| miette::miette!("root block missing from CAR"))?; 334 335 + let root_commit = 336 + jacquard_repo::commit::Commit::from_cbor(&root_bytes).into_diagnostic()?; 337 + debug!( 338 + "backfilling repo at revision {}, root cid {}", 339 + root_commit.rev, root_commit.data 340 + ); 341 342 // 4.5. verify commit signature 343 if verify_signatures { 344 let pubkey = app_state.resolver.resolve_signing_key(did).await?; 345 + root_commit 346 .verify(&pubkey) 347 .map_err(|e| miette::miette!("signature verification failed for {did}: {e}"))?; 348 trace!("signature verified for {did}"); ··· 350 351 // 5. walk mst 352 let start = Instant::now(); 353 + let mst: Mst<MemoryBlockStore> = Mst::load(store, root_commit.data, None); 354 let leaves = mst.leaves().await.into_diagnostic()?; 355 trace!("walked mst for {} in {:?}", did, start.elapsed()); 356 ··· 359 let (_state, added_records, added_blocks, count) = { 360 let app_state = app_state.clone(); 361 let did = did.clone(); 362 + let rev = root_commit.rev; 0 363 364 tokio::task::spawn_blocking(move || { 365 let mut count = 0; ··· 367 let mut added_blocks = 0; 368 let mut collection_counts: HashMap<SmolStr, u64> = HashMap::new(); 369 let mut batch = app_state.db.inner.batch(); 370 + let store = mst.storage(); 371 372 // pre-load existing record CIDs for this DID to detect duplicates/updates 373 let prefix = keys::record_prefix(&did); ··· 383 384 for (key, cid) in leaves { 385 let val_bytes = tokio::runtime::Handle::current() 386 + .block_on(store.get(&cid)) 387 .into_diagnostic()?; 388 389 if let Some(val) = val_bytes { ··· 396 let (action, is_new) = if let Some(existing_cid) = existing_cids.get(&path) 397 { 398 if existing_cid == cid.as_str() { 399 + debug!("skip {did}/{collection}/{rkey} ({cid})"); 400 continue; // skip unchanged record 401 } 402 ("update", false) 403 } else { 404 ("create", true) 405 }; 406 + debug!("{action} {did}/{collection}/{rkey} ({cid})"); 407 408 let db_key = keys::record_key(&did, &collection, rkey); 409 ··· 440 // 6. update status to synced 441 state.status = RepoStatus::Synced; 442 state.rev = Some(rev); 443 + state.data = Some(Cid::ipld(root_commit.data)); 444 state.last_updated_at = chrono::Utc::now().timestamp(); 445 446 batch.insert(

+104

src/bin/mst_dump.rs

··· 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0

··· 1 + use std::env; 2 + use std::sync::Arc; 3 + use std::time::Duration; 4 + 5 + use hydrant::resolver::Resolver; 6 + use jacquard::IntoStatic; // Corrected from jacquard_identity::IntoStatic 7 + use jacquard::api::com_atproto::sync::get_repo::GetRepo; 8 + use jacquard::prelude::XrpcExt; 9 + use jacquard::types::did::Did; 10 + use jacquard_common::types::ident::AtIdentifier; 11 + use jacquard_repo::MemoryBlockStore; 12 + use jacquard_repo::mst::Mst; 13 + use miette::{IntoDiagnostic, Result}; 14 + use tracing::{Level, info}; 15 + use tracing_subscriber::FmtSubscriber; // Restored 16 + use url::Url; // Restored 17 + 18 + #[tokio::main] 19 + async fn main() -> Result<()> { 20 + // Setup logging 21 + let subscriber = FmtSubscriber::builder() 22 + .with_max_level(Level::INFO) 23 + .finish(); 24 + tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed"); 25 + 26 + // Parse args 27 + let args: Vec<String> = env::args().collect(); 28 + if args.len() != 2 { 29 + eprintln!("Usage: {} <handle|did>", args[0]); 30 + std::process::exit(1); 31 + } 32 + let identifier_str = &args[1]; 33 + 34 + // Init resolver 35 + let plc_url = Url::parse("https://plc.directory").into_diagnostic()?; 36 + let resolver = Resolver::new(plc_url, 100); 37 + 38 + // Resolve identity 39 + info!("Resolving {}...", identifier_str); 40 + let identifier = if identifier_str.starts_with("did:") { 41 + AtIdentifier::Did(Did::new(identifier_str).map_err(|e| miette::miette!("{}", e))?) 42 + } else { 43 + AtIdentifier::Handle(identifier_str.parse().into_diagnostic()?) 44 + }; 45 + 46 + let did = match identifier { 47 + AtIdentifier::Did(d) => d.into_static(), 48 + AtIdentifier::Handle(h) => { 49 + let d = resolver.resolve_did(&AtIdentifier::Handle(h)).await?; 50 + d 51 + } 52 + }; 53 + info!("Resolved to DID: {}", did); 54 + 55 + let (pds_url, _) = resolver.resolve_identity_info(&did).await?; 56 + info!("PDS URL: {}", pds_url); 57 + 58 + // Fetch repo 59 + info!("Fetching repo..."); 60 + let http = reqwest::Client::builder() 61 + .timeout(Duration::from_secs(30)) 62 + .build() 63 + .into_diagnostic()?; 64 + 65 + let req = GetRepo::new().did(did.clone()).build(); 66 + let resp = http.xrpc(pds_url).send(&req).await.into_diagnostic()?; 67 + let car_bytes = resp.into_output().map_err(|e| miette::miette!("{}", e))?; // explicit map_err 68 + 69 + info!("Fetched {} bytes", car_bytes.body.len()); 70 + 71 + // Parse CAR 72 + let parsed = jacquard_repo::car::reader::parse_car_bytes(&car_bytes.body) 73 + .await 74 + .into_diagnostic()?; 75 + 76 + let store = Arc::new(MemoryBlockStore::new()); 77 + for (_cid, bytes) in &parsed.blocks { 78 + jacquard_repo::BlockStore::put(store.as_ref(), bytes) 79 + .await 80 + .into_diagnostic()?; 81 + } 82 + 83 + // Load MST 84 + let root_bytes = parsed 85 + .blocks 86 + .get(&parsed.root) 87 + .ok_or_else(|| miette::miette!("root block missing from CAR"))?; 88 + 89 + let root_commit = jacquard_repo::commit::Commit::from_cbor(root_bytes).into_diagnostic()?; 90 + info!("Repo rev: {}", root_commit.rev); 91 + 92 + let mst: Mst<MemoryBlockStore> = Mst::load(store.clone(), root_commit.data, None); 93 + let leaves = mst.leaves().await.into_diagnostic()?; 94 + let root = mst.root().await.into_diagnostic()?; 95 + 96 + info!("Found {} records", leaves.len()); 97 + 98 + println!("root -> {}", root); 99 + for (key, cid) in leaves { 100 + println!("{} -> {}", key, cid); 101 + } 102 + 103 + Ok(()) 104 + }

+4 -4

src/ingest/worker.rs

··· 262 return Ok(ProcessResult::Ok); 263 } 264 265 - if let (Some(prev_repo), Some(prev_commit)) = (&repo_state.data, &commit.prev_data) 266 - && prev_repo != &prev_commit.0 267 { 268 warn!( 269 - "gap detected for {}: prev {} != stored {}. triggering backfill", 270 - did, prev_repo, prev_commit.0 271 ); 272 273 let mut batch = state.db.inner.batch();

··· 262 return Ok(ProcessResult::Ok); 263 } 264 265 + if let (Some(repo), Some(prev_commit)) = (&repo_state.data, &commit.prev_data) 266 + && repo != &prev_commit.0 267 { 268 warn!( 269 + "gap detected for {}: repo {} != commit prev {}. triggering backfill", 270 + did, repo, prev_commit.0 271 ); 272 273 let mut batch = state.db.inner.batch();

+10

src/lib.rs

··· 0 0 0 0 0 0 0 0 0 0

··· 1 + pub mod api; 2 + pub mod backfill; 3 + pub mod config; 4 + pub mod crawler; 5 + pub mod db; 6 + pub mod ingest; 7 + pub mod ops; 8 + pub mod resolver; 9 + pub mod state; 10 + pub mod types;

+9 -20

src/main.rs

··· 1 - mod api; 2 - mod backfill; 3 - mod config; 4 - mod crawler; 5 - mod db; 6 - mod ingest; 7 - mod ops; 8 - mod resolver; 9 - mod state; 10 - mod types; 11 - 12 - use crate::config::{Config, SignatureVerification}; 13 - use crate::crawler::Crawler; 14 - use crate::db::set_firehose_cursor; 15 - use crate::ingest::firehose::FirehoseIngestor; 16 - use crate::state::AppState; 17 - use crate::{backfill::BackfillWorker, ingest::worker::FirehoseWorker}; 18 use futures::{FutureExt, TryFutureExt, future::BoxFuture}; 19 use miette::IntoDiagnostic; 20 use mimalloc::MiMalloc; ··· 81 82 if let Err(e) = spawn_blocking({ 83 let state = state.clone(); 84 - move || crate::backfill::manager::queue_pending_backfills(&state) 85 }) 86 .await 87 .into_diagnostic()? ··· 92 93 if let Err(e) = spawn_blocking({ 94 let state = state.clone(); 95 - move || crate::backfill::manager::queue_gone_backfills(&state) 96 }) 97 .await 98 .into_diagnostic()? ··· 103 104 std::thread::spawn({ 105 let state = state.clone(); 106 - move || crate::backfill::manager::retry_worker(state) 107 }); 108 109 tokio::spawn({

··· 1 + use hydrant::config::{Config, SignatureVerification}; 2 + use hydrant::crawler::Crawler; 3 + use hydrant::db::{self, set_firehose_cursor}; 4 + use hydrant::ingest::firehose::FirehoseIngestor; 5 + use hydrant::state::AppState; 6 + use hydrant::{api, backfill::BackfillWorker, ingest::worker::FirehoseWorker}; 0 0 0 0 0 0 0 0 0 0 0 7 use futures::{FutureExt, TryFutureExt, future::BoxFuture}; 8 use miette::IntoDiagnostic; 9 use mimalloc::MiMalloc; ··· 70 71 if let Err(e) = spawn_blocking({ 72 let state = state.clone(); 73 + move || hydrant::backfill::manager::queue_pending_backfills(&state) 74 }) 75 .await 76 .into_diagnostic()? ··· 81 82 if let Err(e) = spawn_blocking({ 83 let state = state.clone(); 84 + move || hydrant::backfill::manager::queue_gone_backfills(&state) 85 }) 86 .await 87 .into_diagnostic()? ··· 92 93 std::thread::spawn({ 94 let state = state.clone(); 95 + move || hydrant::backfill::manager::retry_worker(state) 96 }); 97 98 tokio::spawn({

+8 -7

src/ops.rs

··· 190 191 trace!("parsed car for {did} in {:?}", start.elapsed()); 192 0 0 0 0 0 0 0 193 if let Some(key) = signing_key { 194 - let root_bytes = parsed 195 - .blocks 196 - .get(&parsed.root) 197 - .ok_or_else(|| miette::miette!("root block missing from CAR"))?; 198 - 199 - let repo_commit = jacquard_repo::commit::Commit::from_cbor(root_bytes).into_diagnostic()?; 200 repo_commit 201 .verify(key) 202 .map_err(|e| miette::miette!("signature verification failed for {did}: {e}"))?; ··· 204 } 205 206 repo_state.rev = Some(commit.rev.clone()); 207 - repo_state.data = Some(Cid::ipld(parsed.root)); 208 repo_state.last_updated_at = chrono::Utc::now().timestamp(); 209 210 batch.insert(&db.repos, keys::repo_key(did), ser_repo_state(&repo_state)?);

··· 190 191 trace!("parsed car for {did} in {:?}", start.elapsed()); 192 193 + let root_bytes = parsed 194 + .blocks 195 + .get(&parsed.root) 196 + .ok_or_else(|| miette::miette!("root block missing from CAR"))?; 197 + 198 + let repo_commit = jacquard_repo::commit::Commit::from_cbor(root_bytes).into_diagnostic()?; 199 + 200 if let Some(key) = signing_key { 0 0 0 0 0 0 201 repo_commit 202 .verify(key) 203 .map_err(|e| miette::miette!("signature verification failed for {did}: {e}"))?; ··· 205 } 206 207 repo_state.rev = Some(commit.rev.clone()); 208 + repo_state.data = Some(Cid::ipld(repo_commit.data)); 209 repo_state.last_updated_at = chrono::Utc::now().timestamp(); 210 211 batch.insert(&db.repos, keys::repo_key(did), ser_repo_state(&repo_state)?);