filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Database schema optimization - part 13 (fixes #1882 fixes #2415)

Browse Source
This commit is contained in:
mdecimus
2025-11-25 12:14:51 +01:00
parent c7fc16d9a2
commit 2b614aa536
62 changed files with 1661 additions and 2264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
crates/store/src/write/blob.rs
View File

@@ -6,14 +6,13 @@
use super::{BlobOp, Operation, ValueClass, ValueOp, key::DeserializeBigEndian, now};
use crate::{
BlobStore, Deserialize, IterateParams, Store, U32_LEN, U64_LEN, ValueKey, write::BatchBuilder,
BlobStore, IterateParams, Store, U32_LEN, U64_LEN, ValueKey,
write::{BatchBuilder, BlobLink},
};
use ahash::AHashSet;
use trc::AddContext;
use types::{
blob::BlobClass,
blob_hash::{BLOB_HASH_LEN, BlobHash},
collection::Collection,
};
#[derive(Debug, PartialEq, Eq)]
@@ -42,7 +41,7 @@ impl Store {
account_id,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Reserve {
class: ValueClass::Blob(BlobOp::Quota {
hash: BlobHash::default(),
until: 0,
}),
@@ -51,9 +50,9 @@ impl Store {
account_id: account_id + 1,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Reserve {
class: ValueClass::Blob(BlobOp::Quota {
hash: BlobHash::default(),
until: 0,
until: u64::MAX,
}),
};
@@ -64,8 +63,8 @@ impl Store {
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
let until = key.deserialize_be_u64(key.len() - U64_LEN)?;
if until > now && value.len() == U32_LEN {
let bytes = u32::deserialize(value)?;
if until > now {
let bytes = value.deserialize_be_u32(0)?;
if bytes > 0 {
quota.bytes += bytes as usize;
quota.count += 1;
@@ -93,9 +92,9 @@ impl Store {
account_id: *account_id,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Reserve {
class: ValueClass::Blob(BlobOp::Link {
hash: hash.as_ref().clone(),
until: *expires,
to: BlobLink::Temporary { until: *expires },
}),
},
BlobClass::Linked {
@@ -108,6 +107,7 @@ impl Store {
document_id: *document_id,
class: ValueClass::Blob(BlobOp::Link {
hash: hash.as_ref().clone(),
to: BlobLink::Document,
}),
},
_ => return Ok(false),
@@ -117,54 +117,12 @@ impl Store {
}
pub async fn purge_blobs(&self, blob_store: BlobStore) -> trc::Result<()> {
// Remove expired temporary blobs
let from_key = ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Reserve {
until: 0,
hash: BlobHash::default(),
}),
};
let to_key = ValueKey {
account_id: u32::MAX,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Reserve {
until: 0,
hash: BlobHash::default(),
}),
};
let mut delete_keys = Vec::new();
let mut active_hashes = AHashSet::new();
let now = now();
self.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
let hash = BlobHash::try_from_hash_slice(
key.get(U32_LEN..U32_LEN + BLOB_HASH_LEN)
.ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?,
)
.unwrap();
let until = key.deserialize_be_u64(key.len() - U64_LEN)?;
if until <= now {
delete_keys.push((key.deserialize_be_u32(0)?, BlobOp::Reserve { until, hash }));
} else {
active_hashes.insert(hash);
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
// Validate linked blobs
let from_key = ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Link {
class: ValueClass::Blob(BlobOp::Commit {
hash: BlobHash::default(),
}),
};
@@ -174,9 +132,16 @@ impl Store {
document_id: u32::MAX,
class: ValueClass::Blob(BlobOp::Link {
hash: BlobHash::new_max(),
to: BlobLink::Document,
}),
};
const TEMP_LINK: usize = BLOB_HASH_LEN + U32_LEN + U64_LEN;
const DOC_LINK: usize = BLOB_HASH_LEN + U64_LEN + 1;
let mut last_hash = BlobHash::default();
let mut last_hash_is_linked = true; // Avoid deleting non-existing last_hash on first iteration
let mut delete_keys = Vec::new();
let now = now();
self.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
@@ -185,15 +150,64 @@ impl Store {
.ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?,
)
.unwrap();
let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?;
if document_id != u32::MAX {
if last_hash != hash {
if last_hash != hash {
if !last_hash_is_linked {
delete_keys.push((
None,
BlobOp::Commit {
hash: std::mem::replace(&mut last_hash, hash),
},
));
} else {
last_hash = hash;
}
} else if last_hash != hash && !active_hashes.contains(&hash) {
// Unlinked or expired blob, delete.
delete_keys.push((0, BlobOp::Commit { hash }));
last_hash_is_linked = false;
}
match key.len() {
BLOB_HASH_LEN => {
// Main blob entry
}
TEMP_LINK => {
// Temporary link
let until = key.deserialize_be_u64(BLOB_HASH_LEN + U32_LEN)?;
if until <= now {
let account_id = key.deserialize_be_u32(BLOB_HASH_LEN)?;
delete_keys.push((
Some(account_id),
BlobOp::Link {
hash: last_hash.clone(),
to: BlobLink::Temporary { until },
},
));
if account_id != u32::MAX {
delete_keys.push((
Some(account_id),
BlobOp::Quota {
hash: last_hash.clone(),
until,
},
));
delete_keys.push((
Some(account_id),
BlobOp::Undelete {
hash: last_hash.clone(),
until,
},
));
}
} else {
last_hash_is_linked = true;
}
}
DOC_LINK => {
// Document link
last_hash_is_linked = true;
}
_ => {
return Err(trc::Error::corrupted_key(key, None, trc::location!()));
}
}
Ok(true)
@@ -202,6 +216,10 @@ impl Store {
.await
.caused_by(trc::location!())?;
if !last_hash_is_linked {
delete_keys.push((None, BlobOp::Commit { hash: last_hash }));
}
// Delete expired or unlinked blobs
for (_, op) in &delete_keys {
if let BlobOp::Commit { hash } = op {
@@ -214,96 +232,18 @@ impl Store {
// Delete hashes
let mut batch = BatchBuilder::new();
let mut last_account_id = u32::MAX;
for (account_id, op) in delete_keys.into_iter() {
if batch.is_large_batch() {
last_account_id = u32::MAX;
self.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
if matches!(op, BlobOp::Reserve { .. }) && account_id != last_account_id {
batch.with_account_id(account_id);
last_account_id = account_id;
}
batch.any_op(Operation::Value {
class: ValueClass::Blob(op),
op: ValueOp::Clear,
});
}
if !batch.is_empty() {
self.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
Ok(())
}
pub async fn blob_hash_unlink_account(&self, account_id: u32) -> trc::Result<()> {
// Validate linked blobs
let from_key = ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Link {
hash: BlobHash::default(),
}),
};
let to_key = ValueKey {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::Blob(BlobOp::Link {
hash: BlobHash::new_max(),
}),
};
let mut delete_keys = Vec::new();
self.iterate(
IterateParams::new(from_key, to_key).ascending().no_values(),
|key, _| {
let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?;
if document_id != u32::MAX && key.deserialize_be_u32(BLOB_HASH_LEN)? == account_id {
delete_keys.push((
Collection::from(key[BLOB_HASH_LEN + U32_LEN]),
document_id,
BlobOp::Link {
hash: BlobHash::try_from_hash_slice(
key.get(0..BLOB_HASH_LEN).ok_or_else(|| {
trc::Error::corrupted_key(key, None, trc::location!())
})?,
)
.unwrap(),
},
));
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
// Unlink blobs
let mut batch = BatchBuilder::new();
batch.with_account_id(account_id);
let mut last_collection = Collection::None;
for (collection, document_id, op) in delete_keys.into_iter() {
for (account_id, op) in delete_keys {
if batch.is_large_batch() {
self.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
if let Some(account_id) = account_id {
batch.with_account_id(account_id);
last_collection = Collection::None;
}
if collection != last_collection {
batch.with_collection(collection);
last_collection = collection;
}
batch.with_document(document_id);
batch.any_op(Operation::Value {
class: ValueClass::Blob(op),
op: ValueOp::Clear,

58
crates/store/src/write/key.rs
View File

@@ -9,14 +9,14 @@ use super::{
TaskQueueClass, TelemetryClass, ValueClass,
};
use crate::{
Deserialize, IndexKey, IndexKeyPrefix, Key, LogKey, SUBSPACE_ACL, SUBSPACE_BLOB_LINK,
SUBSPACE_BLOB_RESERVE, SUBSPACE_COUNTER, SUBSPACE_DIRECTORY, SUBSPACE_IN_MEMORY_COUNTER,
Deserialize, IndexKey, IndexKeyPrefix, Key, LogKey, SUBSPACE_ACL, SUBSPACE_BLOB_EXTRA,
SUBSPACE_BLOB_LINK, SUBSPACE_COUNTER, SUBSPACE_DIRECTORY, SUBSPACE_IN_MEMORY_COUNTER,
SUBSPACE_IN_MEMORY_VALUE, SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_PROPERTY,
SUBSPACE_QUEUE_EVENT, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUOTA, SUBSPACE_REPORT_IN,
SUBSPACE_REPORT_OUT, SUBSPACE_SEARCH_INDEX, SUBSPACE_SETTINGS, SUBSPACE_TASK_QUEUE,
SUBSPACE_TELEMETRY_METRIC, SUBSPACE_TELEMETRY_SPAN, U16_LEN, U32_LEN, U64_LEN, ValueKey,
WITH_SUBSPACE,
write::{IndexPropertyClass, SearchIndex, SearchIndexId, SearchIndexType},
write::{BlobLink, IndexPropertyClass, SearchIndex, SearchIndexId, SearchIndexType},
};
use std::convert::TryInto;
use types::{blob_hash::BLOB_HASH_LEN, collection::SyncCollection};
@@ -329,25 +329,29 @@ impl ValueClass {
.write(document_id),
},
ValueClass::Blob(op) => match op {
BlobOp::Reserve { hash, until } => serializer
BlobOp::Commit { hash } => serializer.write::<&[u8]>(hash.as_ref()),
BlobOp::Link { hash, to } => match to {
BlobLink::Id { id } => serializer.write::<&[u8]>(hash.as_ref()).write(*id),
BlobLink::Document => serializer
.write::<&[u8]>(hash.as_ref())
.write(account_id)
.write(collection)
.write(document_id),
BlobLink::Temporary { until } => serializer
.write::<&[u8]>(hash.as_ref())
.write(account_id)
.write(*until),
},
BlobOp::Quota { hash, until } => serializer
.write(account_id)
.write(0u8)
.write::<&[u8]>(hash.as_ref())
.write(*until),
BlobOp::Commit { hash } => serializer
.write::<&[u8]>(hash.as_ref())
.write(u32::MAX)
.write(0u8)
.write(u32::MAX),
BlobOp::Link { hash } => serializer
.write::<&[u8]>(hash.as_ref())
BlobOp::Undelete { hash, until } => serializer
.write(account_id)
.write(collection)
.write(document_id),
BlobOp::LinkId { hash, id } => serializer
.write(1u8)
.write::<&[u8]>(hash.as_ref())
.write((*id >> 32) as u32)
.write(u8::MAX)
.write(*id as u32),
.write(*until),
},
ValueClass::Config(key) => serializer.write(key.as_slice()),
ValueClass::InMemory(lookup) => match lookup {
@@ -561,9 +565,17 @@ impl ValueClass {
DirectoryClass::Index { word, .. } => word.len() + U32_LEN,
},
ValueClass::Blob(op) => match op {
BlobOp::Reserve { .. } => BLOB_HASH_LEN + U64_LEN + U32_LEN + 1,
BlobOp::Commit { .. } | BlobOp::Link { .. } | BlobOp::LinkId { .. } => {
BLOB_HASH_LEN + U32_LEN * 2 + 2
BlobOp::Commit { .. } => BLOB_HASH_LEN,
BlobOp::Link { to, .. } => {
BLOB_HASH_LEN
+ match to {
BlobLink::Id { .. } => U64_LEN,
BlobLink::Document => U32_LEN * 2 + 1,
BlobLink::Temporary { .. } => U32_LEN + U64_LEN,
}
}
BlobOp::Quota { .. } | BlobOp::Undelete { .. } => {
BLOB_HASH_LEN + U32_LEN + U64_LEN + 1
}
},
ValueClass::TaskQueue(e) => match e {
@@ -624,10 +636,8 @@ impl ValueClass {
ValueClass::Acl(_) => SUBSPACE_ACL,
ValueClass::TaskQueue { .. } => SUBSPACE_TASK_QUEUE,
ValueClass::Blob(op) => match op {
BlobOp::Reserve { .. } => SUBSPACE_BLOB_RESERVE,
BlobOp::Commit { .. } | BlobOp::Link { .. } | BlobOp::LinkId { .. } => {
SUBSPACE_BLOB_LINK
}
BlobOp::Commit { .. } | BlobOp::Link { .. } => SUBSPACE_BLOB_LINK,
BlobOp::Quota { .. } | BlobOp::Undelete { .. } => SUBSPACE_BLOB_EXTRA,
},
ValueClass::Config(_) => SUBSPACE_SETTINGS,
ValueClass::InMemory(lookup) => match lookup {

13
crates/store/src/write/mod.rs
View File

@@ -375,10 +375,17 @@ pub struct SetOperation {
#[derive(Debug, PartialEq, Clone, Eq, Hash)]
pub enum BlobOp {
Reserve { hash: BlobHash, until: u64 },
Commit { hash: BlobHash },
Link { hash: BlobHash },
LinkId { hash: BlobHash, id: u64 },
Link { hash: BlobHash, to: BlobLink },
Quota { hash: BlobHash, until: u64 },
Undelete { hash: BlobHash, until: u64 },
}
#[derive(Debug, PartialEq, Clone, Eq, Hash)]
pub enum BlobLink {
Id { id: u64 },
Document,
Temporary { until: u64 },
}
#[derive(Debug, PartialEq, Clone, Eq, Hash)]