TudbuT/conduit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

split migrations function

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2024-06-09 20:50:40 +00:00
parent f52acd9cdf
commit f0557e3303
1 changed files with 641 additions and 578 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

401
src/service/globals/migrations.rs
View file

@ -16,7 +16,7 @@ use ruma::{
}; };
use tracing::{debug, error, info, warn}; use tracing::{debug, error, info, warn};
use crate::{globals::data::Data, services, utils, Config, Error, Result}; use crate::{services, utils, Config, Error, Result};
pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result<()> { pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result<()> {
// Matrix resource ownership is based on the server name; changing it // Matrix resource ownership is based on the server name; changing it
@ -44,6 +44,165 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
if services().users.count()? > 0 { if services().users.count()? > 0 {
// MIGRATIONS // MIGRATIONS
if services().globals.database_version()? < 1 { if services().globals.database_version()? < 1 {
db_lt_1(db, config).await?;
}
if services().globals.database_version()? < 2 {
db_lt_2(db, config).await?;
}
if services().globals.database_version()? < 3 {
db_lt_3(db, config).await?;
}
if services().globals.database_version()? < 4 {
db_lt_4(db, config).await?;
}
if services().globals.database_version()? < 5 {
db_lt_5(db, config).await?;
}
if services().globals.database_version()? < 6 {
db_lt_6(db, config).await?;
}
if services().globals.database_version()? < 7 {
db_lt_7(db, config).await?;
}
if services().globals.database_version()? < 8 {
db_lt_8(db, config).await?;
}
if services().globals.database_version()? < 9 {
db_lt_9(db, config).await?;
}
if services().globals.database_version()? < 10 {
db_lt_10(db, config).await?;
}
if services().globals.database_version()? < 11 {
db_lt_11(db, config).await?;
}
if services().globals.database_version()? < 12 {
db_lt_12(db, config).await?;
}
// This migration can be reused as-is anytime the server-default rules are
// updated.
if services().globals.database_version()? < 13 {
db_lt_13(db, config).await?;
}
#[cfg(feature = "sha256_media")]
if services().globals.database_version()? < 14 && cfg!(feature = "sha256_media") {
feat_sha256_media(db, config).await?;
}
if db
.global
.get(b"fix_bad_double_separator_in_state_cache")?
.is_none()
{
fix_bad_double_separator_in_state_cache(db, config).await?;
}
if db
.global
.get(b"retroactively_fix_bad_data_from_roomuserid_joined")?
.is_none()
{
retroactively_fix_bad_data_from_roomuserid_joined(db, config).await?;
}
assert_eq!(
services().globals.database_version().unwrap(),
latest_database_version,
"Failed asserting local database version {} is equal to known latest conduwuit database version {}",
services().globals.database_version().unwrap(),
latest_database_version
);
{
let patterns = services().globals.forbidden_usernames();
if !patterns.is_empty() {
for user_id in services()
.users
.iter()
.filter_map(Result::ok)
.filter(|user| !services().users.is_deactivated(user).unwrap_or(true))
.filter(|user| user.server_name() == config.server_name)
{
let matches = patterns.matches(user_id.localpart());
if matches.matched_any() {
warn!(
"User {} matches the following forbidden username patterns: {}",
user_id.to_string(),
matches
.into_iter()
.map(|x| &patterns.patterns()[x])
.join(", ")
);
}
}
}
}
{
let patterns = services().globals.forbidden_alias_names();
if !patterns.is_empty() {
for address in services().rooms.metadata.iter_ids() {
let room_id = address?;
let room_aliases = services().rooms.alias.local_aliases_for_room(&room_id);
for room_alias_result in room_aliases {
let room_alias = room_alias_result?;
let matches = patterns.matches(room_alias.alias());
if matches.matched_any() {
warn!(
"Room with alias {} ({}) matches the following forbidden room name patterns: {}",
room_alias,
&room_id,
matches
.into_iter()
.map(|x| &patterns.patterns()[x])
.join(", ")
);
}
}
}
}
}
info!(
"Loaded {} database with schema version {}",
config.database_backend, latest_database_version
);
} else {
services()
.globals
.bump_database_version(latest_database_version)?;
db.global
.insert(b"fix_bad_double_separator_in_state_cache", &[])?;
db.global
.insert(b"retroactively_fix_bad_data_from_roomuserid_joined", &[])?;
// Create the admin room and server user on first run
services().admin.create_admin_room().await?;
warn!(
"Created new {} database with version {}",
config.database_backend, latest_database_version
);
}
Ok(())
}
async fn db_lt_1(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
for (roomserverid, _) in db.roomserverids.iter() { for (roomserverid, _) in db.roomserverids.iter() {
let mut parts = roomserverid.split(|&b| b == 0xFF); let mut parts = roomserverid.split(|&b| b == 0xFF);
let room_id = parts.next().expect("split always returns one element"); let room_id = parts.next().expect("split always returns one element");
@ -59,11 +218,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(1)?; services().globals.bump_database_version(1)?;
info!("Migration: 0 -> 1 finished");
Ok(())
}
warn!("Migration: 0 -> 1 finished"); async fn db_lt_2(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 2 {
// We accidentally inserted hashed versions of "" into the db instead of just "" // We accidentally inserted hashed versions of "" into the db instead of just ""
for (userid, password) in db.userid_password.iter() { for (userid, password) in db.userid_password.iter() {
let empty_pass = utils::hash::password("").expect("our own password to be properly hashed"); let empty_pass = utils::hash::password("").expect("our own password to be properly hashed");
@ -75,11 +234,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(2)?; services().globals.bump_database_version(2)?;
info!("Migration: 1 -> 2 finished");
Ok(())
}
warn!("Migration: 1 -> 2 finished"); async fn db_lt_3(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 3 {
// Move media to filesystem // Move media to filesystem
for (key, content) in db.mediaid_file.iter() { for (key, content) in db.mediaid_file.iter() {
if content.is_empty() { if content.is_empty() {
@ -94,11 +253,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(3)?; services().globals.bump_database_version(3)?;
info!("Migration: 2 -> 3 finished");
Ok(())
}
warn!("Migration: 2 -> 3 finished"); async fn db_lt_4(_db: &KeyValueDatabase, config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 4 {
// Add federated users to services() as deactivated // Add federated users to services() as deactivated
for our_user in services().users.iter() { for our_user in services().users.iter() {
let our_user = our_user?; let our_user = our_user?;
@ -117,11 +276,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(4)?; services().globals.bump_database_version(4)?;
info!("Migration: 3 -> 4 finished");
Ok(())
}
warn!("Migration: 3 -> 4 finished"); async fn db_lt_5(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 5 {
// Upgrade user data store // Upgrade user data store
for (roomuserdataid, _) in db.roomuserdataid_accountdata.iter() { for (roomuserdataid, _) in db.roomuserdataid_accountdata.iter() {
let mut parts = roomuserdataid.split(|&b| b == 0xFF); let mut parts = roomuserdataid.split(|&b| b == 0xFF);
@ -140,11 +299,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(5)?; services().globals.bump_database_version(5)?;
info!("Migration: 4 -> 5 finished");
Ok(())
}
warn!("Migration: 4 -> 5 finished"); async fn db_lt_6(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 6 {
// Set room member count // Set room member count
for (roomid, _) in db.roomid_shortstatehash.iter() { for (roomid, _) in db.roomid_shortstatehash.iter() {
let string = utils::string_from_bytes(&roomid).unwrap(); let string = utils::string_from_bytes(&roomid).unwrap();
@ -153,11 +312,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(6)?; services().globals.bump_database_version(6)?;
info!("Migration: 5 -> 6 finished");
Ok(())
}
warn!("Migration: 5 -> 6 finished"); async fn db_lt_7(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 7 {
// Upgrade state store // Upgrade state store
let mut last_roomstates: HashMap<OwnedRoomId, u64> = HashMap::new(); let mut last_roomstates: HashMap<OwnedRoomId, u64> = HashMap::new();
let mut current_sstatehash: Option<u64> = None; let mut current_sstatehash: Option<u64> = None;
@ -276,11 +435,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(7)?; services().globals.bump_database_version(7)?;
info!("Migration: 6 -> 7 finished");
Ok(())
}
warn!("Migration: 6 -> 7 finished"); async fn db_lt_8(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 8 {
// Generate short room ids for all rooms // Generate short room ids for all rooms
for (room_id, _) in db.roomid_shortstatehash.iter() { for (room_id, _) in db.roomid_shortstatehash.iter() {
let shortroomid = services().globals.next_count()?.to_be_bytes(); let shortroomid = services().globals.next_count()?.to_be_bytes();
@ -333,11 +492,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
db.eventid_pduid.insert_batch(&mut batch2)?; db.eventid_pduid.insert_batch(&mut batch2)?;
services().globals.bump_database_version(8)?; services().globals.bump_database_version(8)?;
info!("Migration: 7 -> 8 finished");
Ok(())
}
warn!("Migration: 7 -> 8 finished"); async fn db_lt_9(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 9 {
// Update tokenids db layout // Update tokenids db layout
let mut iter = db let mut iter = db
.tokenids .tokenids
@ -389,11 +548,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(9)?; services().globals.bump_database_version(9)?;
info!("Migration: 8 -> 9 finished");
Ok(())
}
warn!("Migration: 8 -> 9 finished"); async fn db_lt_10(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 10 {
// Add other direction for shortstatekeys // Add other direction for shortstatekeys
for (statekey, shortstatekey) in db.statekey_shortstatekey.iter() { for (statekey, shortstatekey) in db.statekey_shortstatekey.iter() {
db.shortstatekey_statekey db.shortstatekey_statekey
@ -406,20 +565,21 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(10)?; services().globals.bump_database_version(10)?;
info!("Migration: 9 -> 10 finished");
Ok(())
}
warn!("Migration: 9 -> 10 finished"); async fn db_lt_11(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 11 {
db.db db.db
.open_tree("userdevicesessionid_uiaarequest")? .open_tree("userdevicesessionid_uiaarequest")?
.clear()?; .clear()?;
services().globals.bump_database_version(11)?; services().globals.bump_database_version(11)?;
info!("Migration: 10 -> 11 finished");
Ok(())
}
warn!("Migration: 10 -> 11 finished"); async fn db_lt_12(_db: &KeyValueDatabase, config: &Config) -> Result<()> {
}
if services().globals.database_version()? < 12 {
for username in services().users.list_local_users()? { for username in services().users.list_local_users()? {
let user = match UserId::parse_with_server_name(username.clone(), &config.server_name) { let user = match UserId::parse_with_server_name(username.clone(), &config.server_name) {
Ok(u) => u, Ok(u) => u,
@ -483,13 +643,11 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(12)?; services().globals.bump_database_version(12)?;
info!("Migration: 11 -> 12 finished");
Ok(())
}
warn!("Migration: 11 -> 12 finished"); async fn db_lt_13(_db: &KeyValueDatabase, config: &Config) -> Result<()> {
}
// This migration can be reused as-is anytime the server-default rules are
// updated.
if services().globals.database_version()? < 13 {
for username in services().users.list_local_users()? { for username in services().users.list_local_users()? {
let user = match UserId::parse_with_server_name(username.clone(), &config.server_name) { let user = match UserId::parse_with_server_name(username.clone(), &config.server_name) {
Ok(u) => u, Ok(u) => u,
@ -522,14 +680,13 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(13)?; services().globals.bump_database_version(13)?;
info!("Migration: 12 -> 13 finished");
Ok(())
}
warn!("Migration: 12 -> 13 finished"); #[cfg(feature = "sha256_media")]
} async fn feat_sha256_media(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
#[cfg(feature = "sha256_media")]
{
use std::path::PathBuf; use std::path::PathBuf;
if services().globals.database_version()? < 14 && cfg!(feature = "sha256_media") {
warn!("sha256_media feature flag is enabled, migrating legacy base64 file names to sha256 file names"); warn!("sha256_media feature flag is enabled, migrating legacy base64 file names to sha256 file names");
// Move old media files to new names // Move old media files to new names
let mut changes = Vec::<(PathBuf, PathBuf)>::new(); let mut changes = Vec::<(PathBuf, PathBuf)>::new();
@ -548,20 +705,15 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
services().globals.bump_database_version(14)?; services().globals.bump_database_version(14)?;
info!("Migration: 13 -> 14 finished");
Ok(())
}
warn!("Migration: 13 -> 14 finished"); async fn fix_bad_double_separator_in_state_cache(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
}
}
if db
.global
.get(b"fix_bad_double_separator_in_state_cache")?
.is_none()
{
warn!("Fixing bad double separator in state_cache roomuserid_joined"); warn!("Fixing bad double separator in state_cache roomuserid_joined");
let mut iter_count: usize = 0; let mut iter_count: usize = 0;
let _cork = db.cork(); let _cork = db.db.cork();
for (mut key, value) in db.roomuserid_joined.iter() { for (mut key, value) in db.roomuserid_joined.iter() {
iter_count = iter_count.saturating_add(1); iter_count = iter_count.saturating_add(1);
@ -572,7 +724,8 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
.iter() .iter()
.get(first_sep_index..=first_sep_index + 1) .get(first_sep_index..=first_sep_index + 1)
.copied() .copied()
.collect_vec() == vec![0xFF, 0xFF] .collect_vec()
== vec![0xFF, 0xFF]
{ {
debug_warn!("Found bad key: {key:?}"); debug_warn!("Found bad key: {key:?}");
db.roomuserid_joined.remove(&key)?; db.roomuserid_joined.remove(&key)?;
@ -583,19 +736,15 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
} }
} }
db.cleanup()?; db.db.cleanup()?;
warn!("Finished fixing");
db.global db.global
.insert(b"fix_bad_double_separator_in_state_cache", &[])?; .insert(b"fix_bad_double_separator_in_state_cache", &[])?;
}
if db info!("Finished fixing");
.global Ok(())
.get(b"retroactively_fix_bad_data_from_roomuserid_joined")? }
.is_none()
{ async fn retroactively_fix_bad_data_from_roomuserid_joined(db: &KeyValueDatabase, _config: &Config) -> Result<()> {
warn!("Retroactively fixing bad data from broken roomuserid_joined"); warn!("Retroactively fixing bad data from broken roomuserid_joined");
let room_ids = services() let room_ids = services()
@ -605,7 +754,7 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
.filter_map(Result::ok) .filter_map(Result::ok)
.collect_vec(); .collect_vec();
let _cork = db.cork(); let _cork = db.db.cork();
for room_id in room_ids.clone() { for room_id in room_ids.clone() {
debug_info!("Fixing room {room_id}"); debug_info!("Fixing room {room_id}");
@ -638,8 +787,7 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
.get_member(&room_id, user_id) .get_member(&room_id, user_id)
.unwrap_or(None) .unwrap_or(None)
.map_or(false, |membership| { .map_or(false, |membership| {
membership.membership == MembershipState::Leave membership.membership == MembershipState::Leave || membership.membership == MembershipState::Ban
|| membership.membership == MembershipState::Ban
}) })
}) })
.collect_vec(); .collect_vec();
@ -663,101 +811,16 @@ pub(crate) async fn migrations(db: &KeyValueDatabase, config: &Config) -> Result
for room_id in room_ids { for room_id in room_ids {
debug_info!( debug_info!(
"Updating joined count for room {room_id} to fix servers in room after correcting membership \ "Updating joined count for room {room_id} to fix servers in room after correcting membership states"
states"
); );
services().rooms.state_cache.update_joined_count(&room_id)?; services().rooms.state_cache.update_joined_count(&room_id)?;
} }
db.cleanup()?; db.db.cleanup()?;
warn!("Finished fixing");
db.global
.insert(b"retroactively_fix_bad_data_from_roomuserid_joined", &[])?;
}
assert_eq!(
services().globals.database_version().unwrap(),
latest_database_version,
"Failed asserting local database version {} is equal to known latest conduwuit database version {}",
services().globals.database_version().unwrap(),
latest_database_version
);
{
let patterns = services().globals.forbidden_usernames();
if !patterns.is_empty() {
for user_id in services()
.users
.iter()
.filter_map(Result::ok)
.filter(|user| !services().users.is_deactivated(user).unwrap_or(true))
.filter(|user| user.server_name() == config.server_name)
{
let matches = patterns.matches(user_id.localpart());
if matches.matched_any() {
warn!(
"User {} matches the following forbidden username patterns: {}",
user_id.to_string(),
matches
.into_iter()
.map(|x| &patterns.patterns()[x])
.join(", ")
);
}
}
}
}
{
let patterns = services().globals.forbidden_alias_names();
if !patterns.is_empty() {
for address in services().rooms.metadata.iter_ids() {
let room_id = address?;
let room_aliases = services().rooms.alias.local_aliases_for_room(&room_id);
for room_alias_result in room_aliases {
let room_alias = room_alias_result?;
let matches = patterns.matches(room_alias.alias());
if matches.matched_any() {
warn!(
"Room with alias {} ({}) matches the following forbidden room name patterns: {}",
room_alias,
&room_id,
matches
.into_iter()
.map(|x| &patterns.patterns()[x])
.join(", ")
);
}
}
}
}
}
info!(
"Loaded {} database with schema version {}",
config.database_backend, latest_database_version
);
} else {
services()
.globals
.bump_database_version(latest_database_version)?;
db.global
.insert(b"fix_bad_double_separator_in_state_cache", &[])?;
db.global db.global
.insert(b"retroactively_fix_bad_data_from_roomuserid_joined", &[])?; .insert(b"retroactively_fix_bad_data_from_roomuserid_joined", &[])?;
// Create the admin room and server user on first run info!("Finished fixing");
services().admin.create_admin_room().await?;
warn!(
"Created new {} database with version {}",
config.database_backend, latest_database_version
);
}
Ok(()) Ok(())
} }