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

CLI account management + Directory refactoring

Browse Source
This commit is contained in:
mdecimus
2023-12-18 22:25:42 +01:00
parent 566a2a0ab8
commit ea94de6d77
91 changed files with 3531 additions and 1846 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/store/src/backend/elastic/mod.rs
View File

@@ -66,7 +66,7 @@ impl ElasticSearchStore {
if let Some(credentials) = credentials {
builder = builder.auth(credentials);
}
if config.property_or_static::<bool>((&prefix, "allow-invalid-certs"), "false")? {
if config.property_or_static::<bool>((&prefix, "tls.allow-invalid-certs"), "false")? {
builder = builder.cert_validation(CertificateValidation::None);
}

14
crates/store/src/backend/foundationdb/main.rs
View File

@@ -21,6 +21,8 @@
* for more details.
*/
use std::time::Duration;
use foundationdb::{options::DatabaseOption, Database};
use utils::config::{utils::AsKey, Config};
@@ -32,14 +34,18 @@ impl FdbStore {
let guard = unsafe { foundationdb::boot() };
let db = Database::new(config.value((&prefix, "path")))?;
if let Some(value) = config.property((&prefix, "transaction.timeout"))? {
db.set_option(DatabaseOption::TransactionTimeout(value))?;
if let Some(value) = config.property::<Duration>((&prefix, "transaction.timeout"))? {
db.set_option(DatabaseOption::TransactionTimeout(value.as_millis() as i32))?;
}
if let Some(value) = config.property((&prefix, "transaction.retry-limit"))? {
db.set_option(DatabaseOption::TransactionRetryLimit(value))?;
}
if let Some(value) = config.property((&prefix, "transaction.max-retry-delay"))? {
db.set_option(DatabaseOption::TransactionMaxRetryDelay(value))?;
if let Some(value) =
config.property::<Duration>((&prefix, "transaction.max-retry-delay"))?
{
db.set_option(DatabaseOption::TransactionMaxRetryDelay(
value.as_millis() as i32
))?;
}
if let Some(value) = config.property((&prefix, "transaction.machine-id"))? {
db.set_option(DatabaseOption::MachineId(value))?;

22
crates/store/src/backend/fs/mod.rs
View File

@@ -41,17 +41,19 @@ impl FsStore {
pub async fn open(config: &Config, prefix: impl AsKey) -> crate::Result<Self> {
let prefix = prefix.as_key();
let path = config.property_require::<PathBuf>((&prefix, "path"))?;
if path.exists() {
Ok(FsStore {
path,
hash_levels: std::cmp::min(config.property_or_static((&prefix, "depth"), "2")?, 5),
})
} else {
Err(crate::Error::InternalError(format!(
"Blob store path {:?} does not exist",
path
)))
if !path.exists() {
fs::create_dir_all(&path).await.map_err(|e| {
crate::Error::InternalError(format!(
"Failed to create blob store path {:?}: {}",
path, e
))
})?;
}
Ok(FsStore {
path,
hash_levels: std::cmp::min(config.property_or_static((&prefix, "depth"), "2")?, 5),
})
}
pub(crate) async fn get_blob(

4
crates/store/src/backend/redis/mod.rs
View File

@@ -91,10 +91,10 @@ impl RedisStore {
builder = builder.retries(value);
}
if let Some(value) = config.property::<Duration>((&prefix, "max-retry-wait"))? {
builder = builder.max_retry_wait(value.as_secs());
builder = builder.max_retry_wait(value.as_millis() as u64);
}
if let Some(value) = config.property::<Duration>((&prefix, "min-retry-wait"))? {
builder = builder.min_retry_wait(value.as_secs());
builder = builder.min_retry_wait(value.as_millis() as u64);
}
if let Some(true) = config.property::<bool>((&prefix, "read-from-replicas"))? {
builder = builder.read_from_replicas();

2
crates/store/src/backend/rocksdb/main.rs
View File

@@ -99,7 +99,7 @@ impl RocksDbStore {
config
.property::<usize>((&prefix, "pool.workers"))?
.filter(|v| *v > 0)
.unwrap_or_else(num_cpus::get),
.unwrap_or_else(|| num_cpus::get() * 4),
)
.build()
.map_err(|err| {

6
crates/store/src/backend/sqlite/main.rs
View File

@@ -40,7 +40,11 @@ impl SqliteStore {
let prefix = prefix.as_key();
let db = Self {
conn_pool: Pool::builder()
.max_size(config.property_or_static((&prefix, "pool.max-connections"), "10")?)
.max_size(
config
.property((&prefix, "pool.max-connections"))?
.unwrap_or_else(|| (num_cpus::get() * 4) as u32),
)
.build(
SqliteConnectionManager::file(
config

2
crates/store/src/config.rs
View File

@@ -210,7 +210,7 @@ impl ConfigStore for Config {
config.lookup_stores.insert(store_id, lookup_store.clone());
// Run init queries on database
for (_, query) in self.values(("store", id, "init")) {
for (_, query) in self.values(("store", id, "init.execute")) {
if let Err(err) = lookup_store.query::<usize>(query, Vec::new()).await {
tracing::warn!("Failed to initialize store {id:?}: {err}");
}

6
crates/store/src/dispatch/store.rs
View File

@@ -306,7 +306,7 @@ impl Store {
}
#[cfg(feature = "test_mode")]
pub async fn blob_hash_expire_all(&self) {
pub async fn blob_expire_all(&self) {
use crate::{
write::{key::DeserializeBigEndian, BatchBuilder, BlobOp, Operation, ValueOp},
BlobHash, BLOB_HASH_LEN, U64_LEN,
@@ -367,7 +367,7 @@ impl Store {
pub async fn assert_is_empty(&self, blob_store: crate::BlobStore) {
use crate::{SUBSPACE_BLOBS, SUBSPACE_COUNTERS};
self.blob_hash_expire_all().await;
self.blob_expire_all().await;
self.purge_blobs(blob_store).await.unwrap();
self.purge_bitmaps().await.unwrap();
@@ -420,7 +420,7 @@ impl Store {
);
}
SUBSPACE_VALUES
if key[0] >= 6
if key[0] >= 20
|| key.get(1..5).unwrap_or_default() == u32::MAX.to_be_bytes() =>
{
// Ignore lastId counter and ID mappings

15
crates/store/src/fts/index.rs
View File

@@ -44,6 +44,7 @@ use crate::{
};
use super::Field;
pub const TERM_INDEX_VERSION: u8 = 1;
#[derive(Debug)]
pub(crate) struct Text<'x, T: Into<u8> + Display + Clone + std::fmt::Debug> {
@@ -234,14 +235,13 @@ impl Store {
// Write term index
let mut batch = BatchBuilder::new();
let mut term_index = lz4_flex::compress_prepend_size(&serializer.finalize());
term_index.insert(0, TERM_INDEX_VERSION);
batch
.with_account_id(document.account_id)
.with_collection(document.collection)
.update_document(document.document_id)
.set(
ValueClass::TermIndex,
lz4_flex::compress_prepend_size(&serializer.finalize()),
);
.set(ValueClass::TermIndex, term_index);
self.write(batch.build()).await?;
let mut batch = BatchBuilder::new();
batch
@@ -339,7 +339,12 @@ struct TermIndex {
impl Deserialize for TermIndex {
fn deserialize(bytes: &[u8]) -> crate::Result<Self> {
let bytes = lz4_flex::decompress_size_prepended(bytes)
if bytes.first().copied().unwrap_or_default() != TERM_INDEX_VERSION {
return Err(Error::InternalError(
"Unsupported term index version".to_string(),
));
}
let bytes = lz4_flex::decompress_size_prepended(bytes.get(1..).unwrap_or_default())
.map_err(|_| Error::InternalError("Failed to decompress term index".to_string()))?;
let mut ops = Vec::new();

9
crates/store/src/fts/query.rs
View File

@@ -39,6 +39,8 @@ use crate::{
BitmapKey, Deserialize, Error, Store, ValueKey,
};
use super::index::TERM_INDEX_VERSION;
struct State<T: Into<u8> + Display + Clone + std::fmt::Debug> {
pub op: FtsFilter<T>,
pub bm: Option<RoaringBitmap>,
@@ -278,7 +280,12 @@ impl Store {
impl Deserialize for BigramIndex {
fn deserialize(bytes: &[u8]) -> crate::Result<Self> {
let bytes = lz4_flex::decompress_size_prepended(bytes)
if bytes.first().copied().unwrap_or_default() != TERM_INDEX_VERSION {
return Err(Error::InternalError(
"Unsupported term index version".to_string(),
));
}
let bytes = lz4_flex::decompress_size_prepended(bytes.get(1..).unwrap_or_default())
.map_err(|_| Error::InternalError("Failed to decompress term index".to_string()))?;
let (num_items, pos) = bytes.read_leb128::<usize>().ok_or(Error::InternalError(

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

@@ -275,11 +275,11 @@ impl<T: AsRef<ValueClass> + Sync + Send> Key for ValueKey<T> {
.write(self.document_id),
},
ValueClass::Directory(directory) => match directory {
DirectoryClass::NameToId(name) => serializer.write(8u8).write(name.as_slice()),
DirectoryClass::EmailToId(email) => serializer.write(9u8).write(email.as_slice()),
DirectoryClass::Principal(uid) => serializer.write(10u8).write_leb128(*uid),
DirectoryClass::Domain(name) => serializer.write(11u8).write(name.as_slice()),
DirectoryClass::UsedQuota(uid) => serializer.write(12u8).write_leb128(*uid),
DirectoryClass::NameToId(name) => serializer.write(20u8).write(name.as_slice()),
DirectoryClass::EmailToId(email) => serializer.write(21u8).write(email.as_slice()),
DirectoryClass::Principal(uid) => serializer.write(22u8).write_leb128(*uid),
DirectoryClass::Domain(name) => serializer.write(23u8).write(name.as_slice()),
DirectoryClass::UsedQuota(uid) => serializer.write(24u8).write_leb128(*uid),
},
}
.finalize()