[db] keep a refetch allowance per repo
11 unmodified lines121314151617181920212223242524262728293031107 unmodified linesExpand all139140141142143144145146147148743 unmodified linesExpand all89289389489589689789889989490090190290390435 unmodified lines11 unmodified linesuse crate::resolver::MiniDoc;
#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))]pub(crate) mod event;#[cfg(test)]pub(crate) mod v14;pub(crate) mod v15;pub(crate) mod v16;pub(crate) mod v2;pub(crate) mod v4;pub(crate) mod v7;
#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))]pub(crate) use event::*;pub(crate) use v14::*;pub(crate) use v15::RepoState;pub(crate) use v16::*;
/// which record-scoped body copies an operator redaction removes.#[cfg(feature = "indexer")]#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]107 unmodified linesExpand all tracked: true, retry_count: 0, relay_inactive: None, backfilled_rev: None, refetch: None, refetch_deferred: false, } }
/// whether the last car a backfill wrote already has the commit at `rev`743 unmodified linesExpand all index_id: 7, retry_count: 2, relay_inactive: None, backfilled_rev: None, refetch: None, refetch_deferred: false, }; let encoded = rmp_serde::to_vec(&metadata).unwrap(); assert_eq!(crate::util::hex::encode(&encoded), "95c30702c0c0"); assert_eq!(crate::util::hex::encode(&encoded), "97c30702c0c0c0c2"); let decoded: RepoMetadata = rmp_serde::from_slice(&encoded).unwrap(); assert!(decoded.tracked); assert_eq!(decoded.index_id, 7); assert_eq!(decoded.retry_count, 2);35 unmodified lines544 unmodified linesExpand all545546547548549550551552553236 unmodified linesExpand all790791792793793794795794795796797798799800801802803804805806174 unmodified linesExpand all98198298398498298598698798898999099130 unmodified lines544 unmodified linesExpand allmigration-transform:src/db/migration/v12.rs::copy_pending_entry rust:src/db/migration/v12.rs::copies_a_queue_that_has_no_did_order_copy_yetmigration-transform:src/db/migration/v13.rs::recount_pds_account_counts rust:src/db/migration/v13.rs::repairs_drifted_per_pds_counts_from_the_rowsmigration-transform:src/db/migration/v14.rs::repo_metadata_v14 rust:src/db/migration/v14.rs::old_database_opens_at_v14_with_its_metadata_rows_untouchedmigration-transform:src/db/migration/v15.rs::repo_state_identity_resolved_at rust:src/db/migration/v15.rs::every_build_fences_older_ones_outmigration-transform:src/db/migration/v16.rs::repo_metadata_refetch rust:src/db/migration/v16.rs::a_v15_database_opens_at_v16_with_its_metadata_rows_untouchedmigration-transform:src/db/migration/v2.rs::repo_state_root_commit rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundariesmigration-transform:src/db/migration/v3.rs::firehose_source_is_pds rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundariesmigration-transform:src/db/migration/v3.rs::is_known_relay rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundariesmigration-transform:src/db/migration/v4.rs::repo_state_active rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries236 unmodified linesExpand allserde-attribute:src/types/event/stream.rs::StoredEvent::data::serde(skip_serializing_if="StoredData::is_nothing") rust:src/types/event/stream.rs::stored_data_persisted_golden_fixturesserde-attribute:src/types/event/stream.rs::StoredEvent::did::serde(borrow) rust:src/control/stream/indexer.rs::persisted_eight_byte_string_rkey_replays_verbatimserde-attribute:src/types/event/stream.rs::StoredEvent::live::serde(default) rust:src/control/stream/indexer.rs::persisted_eight_byte_string_rkey_replays_verbatimserde-attribute:src/types/event/stream.rs::StoredEvent::serde(bound(deserialize="'i:'de")) rust:src/control/stream/indexer.rs::persisted_eight_byte_string_rkey_replays_verbatimserde-attribute:src/types/v14.rs::RepoMetadata::backfilled_rev::serde(default) rust:src/types/v14.rs::v4_repo_metadata_rows_decode_with_no_retriesserde-attribute:src/types/v14.rs::RepoMetadata::relay_inactive::serde(default) rust:src/types/v14.rs::v4_repo_metadata_rows_decode_with_no_retriesserde-attribute:src/types/v14.rs::RepoMetadata::retry_count::serde(default) rust:src/types/v14.rs::v4_repo_metadata_rows_decode_with_no_retriesserde-attribute:src/types/v15.rs::RepoState::identity_resolved_at::serde(default,skip_serializing_if="Option::is_none") rust:src/types/v15.rs::an_unstamped_row_is_written_as_a_v7_rowserde-attribute:src/types/v15.rs::RepoState::pds::serde(borrow) rust:src/types/v15.rs::v15_persisted_schema_golden_fixtureserde-attribute:src/types/v15.rs::RepoState::serde(bound(deserialize="'i:'de")) rust:src/types/v15.rs::v15_persisted_schema_golden_fixtureserde-attribute:src/types/v15.rs::RepoState::signing_key::serde(borrow) rust:src/types/v15.rs::v15_persisted_schema_golden_fixtureserde-attribute:src/types/v16.rs::RepoMetadata::backfilled_rev::serde(default) rust:src/types/v16.rs::v4_and_v14_repo_metadata_rows_decode_with_no_refetchesserde-attribute:src/types/v16.rs::RepoMetadata::refetch::serde(default) rust:src/types/v16.rs::v4_and_v14_repo_metadata_rows_decode_with_no_refetchesserde-attribute:src/types/v16.rs::RepoMetadata::refetch_deferred::serde(default) rust:src/types/v16.rs::v4_and_v14_repo_metadata_rows_decode_with_no_refetchesserde-attribute:src/types/v16.rs::RepoMetadata::relay_inactive::serde(default) rust:src/types/v16.rs::v4_and_v14_repo_metadata_rows_decode_with_no_refetchesserde-attribute:src/types/v16.rs::RepoMetadata::retry_count::serde(default) rust:src/types/v16.rs::v4_and_v14_repo_metadata_rows_decode_with_no_refetchesserde-attribute:src/types/v2.rs::Commit::sig::serde(with="jacquard_common::serde_bytes_helper") rust:src/types/v2.rs::v2_persisted_schema_golden_fixtureserde-attribute:src/types/v2.rs::RepoState::pds::serde(borrow) rust:src/types/v2.rs::v2_persisted_schema_golden_fixtureserde-attribute:src/types/v2.rs::RepoState::serde(bound(deserialize="'i:'de")) rust:src/types/v2.rs::v2_persisted_schema_golden_fixtureserde-attribute:src/types/v2.rs::RepoState::signing_key::serde(borrow) rust:src/types/v2.rs::v2_persisted_schema_golden_fixture174 unmodified linesExpand allserde-type:src/types/event/stream.rs::MarshallableEvt rust:src/types/event/stream.rs::stream_event_json_envelopes_cover_every_variantserde-type:src/types/event/stream.rs::RecordEvt rust:src/types/event/stream.rs::stream_event_json_envelopes_cover_every_variantserde-type:src/types/event/stream.rs::StoredData rust:src/types/event/stream.rs::stored_data_persisted_golden_fixturesserde-type:src/types/event/stream.rs::StoredEvent rust:src/control/stream/indexer.rs::persisted_eight_byte_string_rkey_replays_verbatimserde-type:src/types/v14.rs::RepoMetadata rust:src/types/v14.rs::v14_repo_metadata_golden_fixturesserde-type:src/types/v15.rs::RepoState rust:src/types/v15.rs::v15_persisted_schema_golden_fixtureserde-type:src/types/v16.rs::Allowance rust:src/types/v16.rs::v16_repo_metadata_golden_fixturesserde-type:src/types/v16.rs::RepoMetadata rust:src/types/v16.rs::v16_repo_metadata_golden_fixturesserde-type:src/types/v2.rs::Commit rust:src/types/v2.rs::v2_persisted_schema_golden_fixtureserde-type:src/types/v2.rs::RepoState rust:src/types/v2.rs::v2_persisted_schema_golden_fixtureserde-type:src/types/v2.rs::RepoStatus rust:src/types/v2.rs::v2_persisted_schema_golden_fixtureserde-type:src/types/v4.rs::RepoMetadata rust:src/types/v4.rs::v4_persisted_schema_golden_fixtures30 unmodified lines11234576 unmodified linespub(crate) use super::v7::{Commit, RepoStatus};use super::v7::RepoStatus;use crate::db::types::DbTid;use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]76 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116pub(crate) use super::v7::{Commit, RepoStatus};use crate::db::types::DbTid;use serde::{Deserialize, Serialize};
/// rows are msgpack arrays, so new fields go last and old rows decode with them defaulted#[derive(Debug, Clone, Serialize, Deserialize)]pub(crate) struct RepoMetadata { pub tracked: bool, pub index_id: u64, /// failed backfills in a row. not on the resync entry, because requeueing deletes that #[serde(default)] pub retry_count: u32, /// last status a relay listing asked the pds about, so the same status isn't asked again #[serde(default)] pub relay_inactive: Option<RepoStatus>, /// a commit at or below this is already in the car, even if the shard gets it after #[serde(default)] pub backfilled_rev: Option<DbTid>, /// a good backfill doesn't reset this, or a peer that keeps sending #sync never runs out #[serde(default)] pub refetch: Option<Allowance>, /// live commits are dropped while this is set, because the refetch replaces what they change #[serde(default)] pub refetch_deferred: bool,}
/// a fixed window that starts at the first use, not a rolling one#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]pub(crate) struct Allowance { pub since: i64, pub used: u32,}
#[cfg(test)]mod tests { use super::*;
#[test] fn v16_repo_metadata_golden_fixtures() { let metadata = RepoMetadata { tracked: true, index_id: 7, retry_count: 3, relay_inactive: Some(RepoStatus::Takendown), backfilled_rev: Some(DbTid::new_from_bytes(0x0102030405060708_u64.to_be_bytes())), refetch: Some(Allowance { since: 1000, used: 2, }), refetch_deferred: true, }; let encoded = rmp_serde::to_vec(&metadata).unwrap(); assert_eq!( crate::util::hex::encode(&encoded), "97c30703a954616b656e646f776ec408010203040506070892cd03e802c3" ); let decoded: RepoMetadata = rmp_serde::from_slice(&encoded).unwrap(); assert!(decoded.tracked); assert_eq!(decoded.index_id, 7); assert_eq!(decoded.retry_count, 3); assert_eq!(decoded.relay_inactive, Some(RepoStatus::Takendown)); assert_eq!(decoded.backfilled_rev, metadata.backfilled_rev); assert_eq!(decoded.refetch, metadata.refetch); assert!(decoded.refetch_deferred);
let healthy = rmp_serde::to_vec(&RepoMetadata { relay_inactive: None, retry_count: 0, backfilled_rev: None, refetch: None, refetch_deferred: false, ..metadata }) .unwrap(); assert_eq!(crate::util::hex::encode(&healthy), "97c30700c0c0c0c2"); }
#[test] fn v4_and_v14_repo_metadata_rows_decode_with_no_refetches() { let v4 = super::super::v4::RepoMetadata { tracked: false, index_id: 7, }; let encoded = rmp_serde::to_vec(&v4).unwrap(); assert_eq!(crate::util::hex::encode(&encoded), "92c207"); let decoded: RepoMetadata = rmp_serde::from_slice(&encoded).unwrap(); assert!(!decoded.tracked); assert_eq!(decoded.index_id, 7); assert_eq!(decoded.retry_count, 0); assert_eq!(decoded.relay_inactive, None); assert_eq!(decoded.backfilled_rev, None); assert_eq!(decoded.refetch, None); assert!(!decoded.refetch_deferred);
let v14 = super::super::v14::RepoMetadata { tracked: true, index_id: 7, retry_count: 3, relay_inactive: Some(RepoStatus::Takendown), backfilled_rev: Some(DbTid::new_from_bytes(0x0102030405060708_u64.to_be_bytes())), }; let decoded: RepoMetadata = rmp_serde::from_slice(&rmp_serde::to_vec(&v14).unwrap()).unwrap(); assert_eq!(decoded.retry_count, 3); assert_eq!(decoded.backfilled_rev, v14.backfilled_rev); assert_eq!(decoded.refetch, None); assert!(!decoded.refetch_deferred); }
#[test] fn a_v14_reader_rejects_a_v16_row() { let encoded = crate::util::hex::decode("97c30700c0c0c0c2").unwrap(); assert!(rmp_serde::from_slice::<RepoMetadata>(&encoded).is_ok()); assert!(rmp_serde::from_slice::<super::super::v14::RepoMetadata>(&encoded).is_err()); }}67 unmodified lines686970717273747576188 unmodified linesExpand all265266267268269270271272273274275276927 unmodified linesExpand all67 unmodified linesmod v12;mod v13;mod v14;mod v15;mod v16;mod v2;mod v3;mod v4;mod v5;188 unmodified linesExpand all ( FENCE, Migration::Atomic(v15::repo_state_identity_resolved_at), ), ( "repo_metadata_refetch", Migration::Atomic(v16::repo_metadata_refetch), ),];
/// from this migration on, the applied markers are the whole truth. a relay-only build skips v8/// and the indexer passes for good, so a contiguous version stays at 7 there and would never927 unmodified linesExpand all77 unmodified lines78798081828283848519 unmodified lines77 unmodified lines db.persist()?; }
let db = Db::open(&cfg(tmp.path()))?; assert_eq!(LATEST_VERSION, 15); assert_eq!(version(&db)?, LATEST_VERSION); let epoch = crate::db::counts::identity_ttl_epoch(&db)?; assert!((chrono::Utc::now().timestamp() - epoch).abs() < 60); let row = db19 unmodified lines1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374//! writes nothing on purpose. the bump makes an older build refuse the db at open instead of//! failing every metadata read. don't drop it as a no-op
use fjall::OwnedWriteBatch;use miette::Result;
use crate::db::Db;
pub(super) fn repo_metadata_refetch(_db: &Db, _batch: &mut OwnedWriteBatch) -> Result<bool> { Ok(true)}
#[cfg(all(test, feature = "indexer"))]mod tests { use crate::config::Config; use crate::db::keys::{self, VERSIONING_KEY}; use crate::db::migration::{LATEST_VERSION, migration_applied_key}; use crate::db::{Db, deser_repo_meta}; use crate::types::v14; use jacquard_common::types::did::Did; use miette::{IntoDiagnostic, Result}; use tempfile::tempdir;
#[test] fn a_v15_database_opens_at_v16_with_its_metadata_rows_untouched() -> Result<()> { let tmp = tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?; let old_row = rmp_serde::to_vec(&v14::RepoMetadata { tracked: true, index_id: 7, retry_count: 2, relay_inactive: None, backfilled_rev: None, }) .into_diagnostic()?; { let db = Db::open(&cfg)?; db.repo_metadata .insert(keys::repo_metadata_key(&did), &old_row) .into_diagnostic()?; let mut batch = db.inner.batch(); batch.remove(&db.counts, migration_applied_key("repo_metadata_refetch")); batch.insert(&db.counts, VERSIONING_KEY, 15_u64.to_be_bytes()); batch.commit().into_diagnostic()?; db.persist()?; }
let db = Db::open(&cfg)?; let version = db .counts .get(VERSIONING_KEY) .into_diagnostic()? .expect("schema version"); assert_eq!( u64::from_be_bytes(version.as_ref().try_into().into_diagnostic()?), LATEST_VERSION ); let row = db .repo_metadata .get(keys::repo_metadata_key(&did)) .into_diagnostic()? .expect("metadata row"); assert_eq!(row.as_ref(), old_row.as_slice()); let metadata = deser_repo_meta(&row)?; assert_eq!(metadata.retry_count, 2); assert_eq!(metadata.refetch, None); assert!(!metadata.refetch_deferred); Ok(()) }}