diff --git a/src/db/migration/mod.rs b/src/db/migration/mod.rs index 4291d92..c7de137 100644 --- a/src/db/migration/mod.rs +++ b/src/db/migration/mod.rs @@ -66,6 +66,7 @@ mod v1; mod v10; mod v11; mod v12; +mod v13; mod v2; mod v3; mod v4; @@ -250,6 +251,10 @@ const MIGRATIONS: &[(&str, Migration)] = &[ finalize: None, }, ), + ( + "recount_pds_account_counts", + Migration::Atomic(v13::recount_pds_account_counts), + ), ]; pub(crate) const LATEST_VERSION: u64 = MIGRATIONS.len() as u64; diff --git a/src/db/migration/v13.rs b/src/db/migration/v13.rs new file mode 100644 index 0000000..619f85b --- /dev/null +++ b/src/db/migration/v13.rs @@ -0,0 +1,104 @@ +//! v13 recounts the per-pds account counts from the repo rows the way v6 does, which repairs +//! a database whose counts drifted from the rows they count + +use fjall::OwnedWriteBatch; +use miette::{Context, IntoDiagnostic, Result}; + +use crate::db::Db; +use crate::types::v7; + +use super::v6::{counted_host, recount_pds_accounts}; + +pub(super) fn recount_pds_account_counts(db: &Db, batch: &mut OwnedWriteBatch) -> Result { + recount_pds_accounts(db, batch, |value| { + let state: v7::RepoState = rmp_serde::from_slice(value) + .into_diagnostic() + .wrap_err("invalid v12 repo state")?; + Ok(counted_host(state.active, state.pds.as_deref())) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + use crate::db::keys; + use crate::db::migration::rewind_version_for_test; + use jacquard_common::CowStr; + use jacquard_common::types::did::Did; + use tempfile::tempdir; + + fn counts_on_disk(db: &Db) -> Result, Vec)>> { + db.counts + .prefix(keys::count_keyspace_key("p|")) + .map(|guard| { + let (key, value) = guard.into_inner().into_diagnostic()?; + Ok((key.to_vec(), value.to_vec())) + }) + .collect() + } + + #[test] + fn repairs_drifted_per_pds_counts_from_the_rows() -> Result<()> { + let tmp = tempdir().into_diagnostic()?; + let cfg = Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + }; + let accounts = |db: &Db, host| db.get_count_sync(&keys::pds_account_count_key(host)); + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + let rows = [ + ("did:web:one.test", "https://pds.example/", true), + ("did:web:two.test", "https://pds.example/", true), + ("did:web:gone.test", "https://gone.example/", false), + ("did:web:also-gone.test", "https://gone.example/", false), + ("did:web:fresh.test", "https://fresh.example/", true), + ]; + for (did, pds, active) in rows { + let row = v7::RepoState { + active, + pds: Some(CowStr::Borrowed(pds)), + ..v7::RepoState::backfilling() + }; + batch.insert( + &db.repos, + keys::repo_key(&Did::new_static(did).into_diagnostic()?), + crate::db::ser_repo_state(&row)?, + ); + } + // what the writers that never counted left behind: one short, one host whose + // rows all went inactive still counted, one never counted at all + for (host, count) in [("pds.example", 1), ("gone.example", 2)] { + crate::db::set_ks_count(&mut batch, &db, &keys::pds_account_count_key(host), count); + } + batch.insert( + &db.counts, + keys::count_delta_key(1_000_000, &keys::pds_account_count_key("gone.example")), + 3_i64.to_be_bytes(), + ); + batch.commit().into_diagnostic()?; + rewind_version_for_test(&db, 9)?; + db.persist()?; + } + + let repaired = { + let db = Db::open(&cfg)?; + assert_eq!(accounts(&db, "pds.example"), 2); + assert_eq!(accounts(&db, "gone.example"), 0); + assert_eq!(accounts(&db, "fresh.example"), 1); + let on_disk = counts_on_disk(&db)?; + rewind_version_for_test(&db, 9)?; + db.persist()?; + on_disk + }; + + let db = Db::open(&cfg)?; + assert_eq!(counts_on_disk(&db)?, repaired); + assert_eq!(accounts(&db, "pds.example"), 2); + assert_eq!(accounts(&db, "gone.example"), 0); + assert_eq!(accounts(&db, "fresh.example"), 1); + Ok(()) + } +} diff --git a/src/db/migration/v6.rs b/src/db/migration/v6.rs index feeb803..a713040 100644 --- a/src/db/migration/v6.rs +++ b/src/db/migration/v6.rs @@ -1,35 +1,41 @@ use std::collections::BTreeMap; +use crate::db::counts::pds_host; use crate::db::keys::{self, COUNT_KS_PREFIX}; use crate::db::{Db, set_ks_count}; use crate::types::v4; use fjall::OwnedWriteBatch; use miette::{Context, IntoDiagnostic, Result}; use smol_str::SmolStr; -use url::Url; pub(crate) fn rebuild_pds_account_counts(db: &Db, batch: &mut OwnedWriteBatch) -> Result { + recount_pds_accounts(db, batch, |value| { + let state: v4::RepoState = rmp_serde::from_slice(value) + .into_diagnostic() + .wrap_err("invalid v5 repo state")?; + Ok(counted_host(state.active, state.pds.as_deref())) + }) +} + +/// the host a repo row with this `active` and `pds` counts one account for +pub(super) fn counted_host(active: bool, pds: Option<&str>) -> Option { + pds.filter(|_| active).and_then(pds_host) +} + +/// replaces every per-pds account count, and its pending deltas, with what `host_of` says +/// the repo rows count for, so a host none of them count for anymore goes to zero +pub(super) fn recount_pds_accounts( + db: &Db, + batch: &mut OwnedWriteBatch, + host_of: impl Fn(&[u8]) -> Result>, +) -> Result { let mut counts: BTreeMap = BTreeMap::new(); for guard in db.repos.iter() { let (_, value) = guard.into_inner().into_diagnostic()?; - let state: v4::RepoState = rmp_serde::from_slice(value.as_ref()) - .into_diagnostic() - .wrap_err("invalid v5 repo state")?; - if !state.active { - continue; + if let Some(host) = host_of(&value)? { + *counts.entry(host).or_insert(0) += 1; } - - let Some(host) = state - .pds - .as_deref() - .and_then(|pds| Url::parse(pds).ok()) - .and_then(|url| url.host_str().map(SmolStr::new)) - else { - continue; - }; - - *counts.entry(host).or_insert(0) += 1; } for guard in db.counts.prefix(COUNT_KS_PREFIX) { diff --git a/tests/representation_inventory.tsv b/tests/representation_inventory.tsv index 8cbd66a..85ca9bc 100644 --- a/tests/representation_inventory.tsv +++ b/tests/representation_inventory.tsv @@ -209,6 +209,7 @@ codec-call:src/db/migration/mod.rs::pass_cursor_key::.to_be_bytes rust:src/db/mi codec-call:src/db/migration/mod.rs::run::.to_be_bytes rust:src/db/migration/mod.rs::a_chunk_that_fails_partway_commits_nothing codec-call:src/db/migration/v10/events.rs::decode_event::rmp_serde::from_slice rust:src/db/migration/v10/events.rs::cid_mismatch_aborts_without_clearing_blocks codec-call:src/db/migration/v11.rs::status::rmp_serde::from_slice rust:src/db/migration/v11.rs::splits_legacy_bans_before_canonicalizing_other_status_aliases +codec-call:src/db/migration/v13.rs::recount_pds_account_counts::rmp_serde::from_slice rust:src/db/migration/v13.rs::repairs_drifted_per_pds_counts_from_the_rows codec-call:src/db/migration/v2.rs::repo_state_root_commit::rmp_serde::from_slice rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries codec-call:src/db/migration/v2.rs::repo_state_root_commit::rmp_serde::to_vec rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries codec-call:src/db/migration/v3.rs::firehose_source_is_pds::rmp_serde::to_vec rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries @@ -463,12 +464,15 @@ migration-transform:src/db/migration/v11.rs::rewrite_ban rust:src/db/migration/v migration-transform:src/db/migration/v11.rs::rewrite_status rust:src/db/migration/v11.rs::splits_legacy_bans_before_canonicalizing_other_status_aliases migration-transform:src/db/migration/v11.rs::status rust:src/db/migration/v11.rs::splits_legacy_bans_before_canonicalizing_other_status_aliases migration-transform:src/db/migration/v12.rs::copy_pending_entry rust:src/db/migration/v12.rs::copies_a_queue_that_has_no_did_order_copy_yet +migration-transform:src/db/migration/v13.rs::recount_pds_account_counts rust:src/db/migration/v13.rs::repairs_drifted_per_pds_counts_from_the_rows migration-transform:src/db/migration/v2.rs::repo_state_root_commit rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries migration-transform:src/db/migration/v3.rs::firehose_source_is_pds rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries migration-transform:src/db/migration/v3.rs::is_known_relay rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries migration-transform:src/db/migration/v4.rs::repo_state_active rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries migration-transform:src/db/migration/v5.rs::pds_meta_layout rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries +migration-transform:src/db/migration/v6.rs::counted_host rust:src/db/migration/v13.rs::repairs_drifted_per_pds_counts_from_the_rows migration-transform:src/db/migration/v6.rs::rebuild_pds_account_counts rust:src/db/migration/mod.rs::v0_database_migrates_all_early_representation_boundaries +migration-transform:src/db/migration/v6.rs::recount_pds_accounts rust:src/db/migration/v13.rs::repairs_drifted_per_pds_counts_from_the_rows migration-transform:src/db/migration/v7.rs::repo_state_event_clocks rust:src/db/migration/v7.rs::repo_state_event_clock_migration_rewrites_v6_repo_states migration-transform:src/db/migration/v8.rs::count_gauge_key_from_repo_key rust:src/db/migration/v8.rs::lifecycle_count_migration_rebuilds_skewed_counts_and_drops_old_deltas migration-transform:src/db/migration/v8.rs::is_lifecycle_count rust:src/db/migration/v8.rs::lifecycle_count_migration_rebuilds_skewed_counts_and_drops_old_deltas