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

JMAP Registry API implementation - part 12

Browse Source
This commit is contained in:
mdecimus
2026-03-09 19:21:42 +01:00
parent 455d3e3beb
commit a747239b61
45 changed files with 1718 additions and 883 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -79,118 +79,117 @@ impl Store {
self.get_value::<()>(key).await.map(|v| v.is_some())
}
pub async fn purge_blobs(&self, blob_store: BlobStore) -> trc::Result<()> {
pub async fn purge_blobs(&self, blob_store: BlobStore, shard_index: u8) -> trc::Result<()> {
let mut total_active = 0;
let mut total_deleted = 0;
let started = Instant::now();
for byte in 0..=u8::MAX {
// Validate linked blobs
let mut from_hash = BlobHash::default();
let mut to_hash = BlobHash::new_max();
from_hash.0[0] = byte;
to_hash.0[0] = byte;
let from_key = ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Commit { hash: from_hash }),
};
let to_key = ValueKey {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::Blob(BlobOp::Link {
hash: to_hash,
to: BlobLink::Document,
}),
};
// Validate linked blobs
let mut from_hash = BlobHash::default();
let mut to_hash = BlobHash::new_max();
from_hash.0[0] = shard_index;
to_hash.0[0] = shard_index;
let from_key = ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Blob(BlobOp::Commit { hash: from_hash }),
};
let to_key = ValueKey {
account_id: u32::MAX,
collection: u8::MAX,
document_id: u32::MAX,
class: ValueClass::Blob(BlobOp::Link {
hash: to_hash,
to: BlobLink::Document,
}),
};
let mut state = BlobPurgeState::new();
self.iterate(
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
let hash =
BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else(
|| trc::Error::corrupted_key(key, value.into(), trc::location!()),
)?)
.unwrap();
let mut state = BlobPurgeState::new();
self.iterate(
IterateParams::new(from_key, to_key).ascending(),
|key, value| {
let hash =
BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else(|| {
trc::Error::corrupted_key(key, value.into(), trc::location!())
})?)
.unwrap();
state.update_hash(hash);
state.process_key(key, value)?;
state.update_hash(hash);
state.process_key(key, value)?;
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
state.finalize(BlobHash::default());
state.finalize(BlobHash::default());
// Delete expired or unlinked blobs
for (_, op) in &state.delete_keys {
if let BlobOp::Commit { hash } = op {
blob_store
.delete_blob(hash.as_ref())
.await
.caused_by(trc::location!())?;
}
}
// Delete hashes
let mut batch = BatchBuilder::new();
for (account_id, op) in state.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);
}
batch.any_op(Operation::Value {
class: ValueClass::Blob(op),
op: ValueOp::Clear,
});
}
for (account_id, object_id) in state.delete_registry {
if batch.is_large_batch() {
self.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
let item_id = object_id.id().id();
let object_id = object_id.object().to_id();
batch
.clear(ValueClass::Registry(RegistryClass::Index {
index_id: Property::AccountId.to_id(),
object_id,
item_id,
key: (account_id as u64).serialize(),
}))
.clear(ValueClass::Registry(RegistryClass::Item {
object_id,
item_id,
}));
}
if !batch.is_empty() {
self.write(batch.build_all())
// Delete expired or unlinked blobs
for (_, op) in &state.delete_keys {
if let BlobOp::Commit { hash } = op {
blob_store
.delete_blob(hash.as_ref())
.await
.caused_by(trc::location!())?;
}
total_active += state.total_active - 1; // Exclude default hash
total_deleted += state.total_deleted;
}
// Delete hashes
let mut batch = BatchBuilder::new();
for (account_id, op) in state.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);
}
batch.any_op(Operation::Value {
class: ValueClass::Blob(op),
op: ValueOp::Clear,
});
}
for (account_id, object_id) in state.delete_registry {
if batch.is_large_batch() {
self.write(batch.build_all())
.await
.caused_by(trc::location!())?;
batch = BatchBuilder::new();
}
let item_id = object_id.id().id();
let object_id = object_id.object().to_id();
batch
.clear(ValueClass::Registry(RegistryClass::Index {
index_id: Property::AccountId.to_id(),
object_id,
item_id,
key: (account_id as u64).serialize(),
}))
.clear(ValueClass::Registry(RegistryClass::Item {
object_id,
item_id,
}));
}
if !batch.is_empty() {
self.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
total_active += state.total_active - 1; // Exclude default hash
total_deleted += state.total_deleted;
trc::event!(
Store(StoreEvent::BlobStorePurged),
Id = shard_index as u16,
Expires = total_deleted,
Total = total_active,
Elapsed = started.elapsed()