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

PostgreSQL backend implementation

Browse Source
This commit is contained in:
mdecimus
2023-11-30 12:43:28 +01:00
parent a1f7ad891a
commit 5f36e1f356
40 changed files with 1917 additions and 615 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
tests/src/imap/mod.rs
View File

@@ -136,8 +136,13 @@ allow-invalid-certs = true
future-release = [ { if = "authenticated-as", ne = "", then = "99999999d"},
{ else = false } ]
[store]
db.path = "{TMP}/sqlite.db"
[store.db]
#path = "{TMP}/sqlite.db"
host = "localhost"
port = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
[store.blob]
type = "local"

11
tests/src/jmap/mod.rs
View File

@@ -134,8 +134,13 @@ allow-invalid-certs = true
future-release = [ { if = "authenticated-as", ne = "", then = "99999999d"},
{ else = false } ]
[store]
db.path = "{TMP}/sqlite.db"
[store.db]
path = "{TMP}/sqlite.db"
host = "localhost"
port = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
[store.blob]
type = "local"
@@ -242,7 +247,7 @@ pub async fn jmap_tests() {
let delete = true;
let mut params = init_jmap_tests(delete).await;
//email_query::test(params.server.clone(), &mut params.client, delete).await;
email_query::test(params.server.clone(), &mut params.client, delete).await;
email_get::test(params.server.clone(), &mut params.client).await;
email_set::test(params.server.clone(), &mut params.client).await;
email_parse::test(params.server.clone(), &mut params.client).await;

35
tests/src/jmap/stress_test.rs
View File

@@ -51,7 +51,10 @@ pub async fn test(server: Arc<JMAP>, mut client: Client) {
async fn email_tests(server: Arc<JMAP>, client: Arc<Client>) {
for pass in 0..NUM_PASSES {
println!("----------------- PASS {} -----------------", pass);
println!(
"----------------- EMAIL STRESS TEST {} -----------------",
pass
);
let mailboxes = Arc::new(vec![
client
.mailbox_create("Stress 1", None::<String>, Role::None)
@@ -97,7 +100,12 @@ async fn email_tests(server: Arc<JMAP>, client: Arc<Client>) {
.await
.unwrap()
.take_id();
//println!("Inserted message {}.", message_id);
/*println!(
"Inserted message {}.",
Id::from_bytes(_message_id.as_bytes())
.unwrap()
.document_id()
);*/
}));
}
@@ -110,7 +118,10 @@ async fn email_tests(server: Arc<JMAP>, client: Arc<Client>) {
let ids = req.send_query_email().await.unwrap().take_ids();
if !ids.is_empty() {
let message_id = &ids[rand::thread_rng().gen_range(0..ids.len())];
//println!("Deleting message {}.", message_id);
/*println!(
"Deleting message {}.",
Id::from_bytes(message_id.as_bytes()).unwrap().document_id()
);*/
match client.email_destroy(message_id).await {
Ok(_) => {
break;
@@ -121,7 +132,7 @@ async fn email_tests(server: Arc<JMAP>, client: Arc<Client>) {
}
SetErrorType::Forbidden => {
// Concurrency issue, try again.
println!("Concurrent update, trying again.");
//println!("Concurrent update, trying again.");
}
_ => {
panic!("Unexpected error: {:?}", err);
@@ -169,7 +180,15 @@ async fn email_tests(server: Arc<JMAP>, client: Arc<Client>) {
if new_mailbox_id != mailbox_id {
/*println!(
"Moving message {} from {} to {}.",
message_id, mailbox_id, new_mailbox_id
Id::from_bytes(message_id.as_bytes())
.unwrap()
.document_id(),
Id::from_bytes(mailbox_id.as_bytes())
.unwrap()
.document_id(),
Id::from_bytes(new_mailbox_id.as_bytes())
.unwrap()
.document_id()
);*/
let mut req = client.build();
req.set_email()
@@ -271,6 +290,8 @@ async fn mailbox_tests(server: Arc<JMAP>, client: Arc<Client>) {
]);
let mut futures = Vec::new();
println!("----------------- MAILBOX STRESS TEST -----------------");
for _ in 0..1000 {
match rand::thread_rng().gen_range(0..=3) {
0 => {
@@ -278,6 +299,7 @@ async fn mailbox_tests(server: Arc<JMAP>, client: Arc<Client>) {
let client = client.clone();
let mailboxes = mailboxes.clone();
futures.push(tokio::spawn(async move {
//println!("Creating mailbox {}.", mailboxes[pos]);
create_mailbox(&client, &mailboxes[pos]).await;
}));
}
@@ -286,6 +308,7 @@ async fn mailbox_tests(server: Arc<JMAP>, client: Arc<Client>) {
1 => {
let client = client.clone();
futures.push(tokio::spawn(async move {
//print!("Querying mailboxes...");
query_mailboxes(&client).await;
}));
}
@@ -301,6 +324,7 @@ async fn mailbox_tests(server: Arc<JMAP>, client: Arc<Client>) {
{
let client = client.clone();
tokio::spawn(async move {
//println!("Deleting mailbox {}.", mailbox_id);
delete_mailbox(&client, &mailbox_id).await;
});
}
@@ -318,6 +342,7 @@ async fn mailbox_tests(server: Arc<JMAP>, client: Arc<Client>) {
if !ids.is_empty() {
let id = ids.swap_remove(rand::thread_rng().gen_range(0..ids.len()));
let sort_order = rand::thread_rng().gen_range(0..100);
//println!("Updating mailbox {}.", id);
client.mailbox_update_sort_order(&id, sort_order).await.ok();
}
}));

101
tests/src/store/assign_id.rs
View File

@@ -31,53 +31,126 @@ use store::{write::BatchBuilder, Store};
pub async fn test(db: Store) {
println!("Running Store ID assignment tests...");
ID_ASSIGNMENT_EXPIRY.store(2, std::sync::atomic::Ordering::Relaxed);
test_0(db.clone()).await;
test_1(db.clone()).await;
test_2(db.clone()).await;
test_3(db).await;
test_3(db.clone()).await;
test_4(db).await;
ID_ASSIGNMENT_EXPIRY.store(60 * 60, std::sync::atomic::Ordering::Relaxed);
}
async fn test_0(db: Store) {
// Test document id assignment
println!("Assigning 1000 ids concurrently...");
ID_ASSIGNMENT_EXPIRY.store(10 * 60 * 60, std::sync::atomic::Ordering::Relaxed);
let mut handles = Vec::new();
let mut assigned_ids = HashSet::new();
// Create 1000 ids concurrently
for _ in 0..1000 {
handles.push({
let db = db.clone();
tokio::spawn(async move { db.assign_document_id(0, u8::MAX).await.unwrap() })
});
}
for handle in handles {
let assigned_id = handle.await.unwrap();
assert!(
assigned_ids.insert(assigned_id),
"already assigned or invalid: {assigned_id}"
);
}
assert_eq!(assigned_ids.len(), 1000);
db.destroy().await;
}
async fn test_1(db: Store) {
// Test document id assignment
ID_ASSIGNMENT_EXPIRY.store(2, std::sync::atomic::Ordering::Relaxed);
println!("Assigning 100 ids concurrently and reassign after expiration...");
for wait_for_expiry in [true, false] {
let mut handles = Vec::new();
let mut expected_ids = HashSet::new();
let mut assigned_ids = HashSet::new();
// Create 100 ids concurrently
for id in 0..100 {
for _ in 0..100 {
handles.push({
let db = db.clone();
tokio::spawn(async move { db.assign_document_id(0, u8::MAX).await })
tokio::spawn(async move { db.assign_document_id(0, u8::MAX).await.unwrap() })
});
expected_ids.insert(id);
}
for handle in handles {
let assigned_id = handle.await.unwrap().unwrap();
let assigned_id = handle.await.unwrap();
//println!("assigned id: {assigned_id} ({wait_for_expiry})");
assert!(
expected_ids.remove(&assigned_id),
assigned_ids.insert(assigned_id),
"already assigned or invalid: {assigned_id} ({wait_for_expiry})"
);
}
assert_eq!(
expected_ids.len(),
0,
"{expected_ids:?} ({wait_for_expiry})"
assigned_ids.len(),
100,
"{assigned_ids:?} ({wait_for_expiry})"
);
if wait_for_expiry {
tokio::time::sleep(Duration::from_secs(3)).await;
}
}
db.destroy().await;
}
async fn test_2(db: Store) {
// Test document id assignment
let mut handles = Vec::new();
let mut assigned_ids = HashSet::new();
// Create 1000 ids concurrently
println!("Create 1000 documentIds concurrently...");
ID_ASSIGNMENT_EXPIRY.store(10 * 60 * 60, std::sync::atomic::Ordering::Relaxed);
for _ in 0..1000 {
handles.push({
let db = db.clone();
tokio::spawn(async move {
{
let id = db.assign_document_id(0, u8::MAX).await.unwrap();
db.write(
BatchBuilder::new()
.with_account_id(0)
.with_collection(u8::MAX)
.create_document(id)
.build_batch(),
)
.await
.unwrap();
id
}
})
});
}
for handle in handles {
let assigned_id = handle.await.unwrap();
assert!(
assigned_ids.insert(assigned_id),
"already assigned or invalid: {assigned_id}"
);
}
assert_eq!(assigned_ids.len(), 1000, "{assigned_ids:?} ");
db.destroy().await;
}
async fn test_3(db: Store) {
// Create document ids and try reassigning
println!("Assigning 100 ids concurrently and try reassigning...");
ID_ASSIGNMENT_EXPIRY.store(2, std::sync::atomic::Ordering::Relaxed);
let mut expected_ids = AHashSet::new();
let mut batch = BatchBuilder::new();
batch.with_account_id(0).with_collection(u8::MAX);
@@ -107,8 +180,10 @@ async fn test_2(db: Store) {
db.destroy().await;
}
async fn test_3(db: Store) {
async fn test_4(db: Store) {
// Try reassigning deleted ids
println!("Create and delete 100 documentIds then try reassigning ids...");
ID_ASSIGNMENT_EXPIRY.store(60 * 60, std::sync::atomic::Ordering::Relaxed);
let mut expected_ids = AHashSet::new();
let mut batch = BatchBuilder::new();
batch.with_account_id(0).with_collection(u8::MAX);

15
tests/src/store/blob.rs
View File

@@ -22,7 +22,7 @@
*/
use store::{
backend::{fs::FsStore, s3::S3Store, sqlite::SqliteStore},
backend::{fs::FsStore, postgres::PostgresStore, s3::S3Store, sqlite::SqliteStore},
write::{blob::BlobQuota, now, BatchBuilder, BlobOp, F_CLEAR},
BlobClass, BlobHash, BlobStore, Store,
};
@@ -47,13 +47,19 @@ path = "{TMP}"
const CONFIG_DB: &str = r#"
[store.db]
path = "{TMP}/db.db?mode=rwc"
host = "localhost"
post = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
"#;
#[tokio::test]
pub async fn blob_tests() {
let temp_dir = TempDir::new("blob_tests", true);
for (store_id, store_cfg) in [("s3", CONFIG_S3), ("fs", CONFIG_LOCAL)] {
/*for (store_id, store_cfg) in [("s3", CONFIG_S3), ("fs", CONFIG_LOCAL)] {
let config =
Config::new(&store_cfg.replace("{TMP}", temp_dir.path.as_path().to_str().unwrap()))
.unwrap();
@@ -66,10 +72,11 @@ pub async fn blob_tests() {
println!("Testing blob store {}...", store_id);
test_store(blob_store_.clone()).await;
}
}*/
// Init store
let store: Store = SqliteStore::open(
//let store: Store = SqliteStore::open(
let store: Store = PostgresStore::open(
//let store: Store = FdbStore::open(
&Config::new(&CONFIG_DB.replace("{TMP}", temp_dir.path.as_path().to_str().unwrap()))
.unwrap(),

30
tests/src/store/mod.rs
View File

@@ -29,28 +29,36 @@ use std::io::Read;
use ::store::Store;
use store::backend::{foundationdb::FdbStore, sqlite::SqliteStore};
use store::backend::{foundationdb::FdbStore, postgres::PostgresStore, sqlite::SqliteStore};
use utils::config::Config;
pub struct TempDir {
pub path: std::path::PathBuf,
}
const CONFIG: &str = r#"
[store.blob]
type = "local"
local.path = "PATH"
[store.db]
#path = "PATH/sqlite.db"
host = "localhost"
post = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
"#;
#[tokio::test]
pub async fn store_tests() {
let insert = true;
let temp_dir = TempDir::new("store_tests", insert);
let config_file = format!(
concat!(
"store.blob.type = \"local\"\n",
"store.blob.local.path = \"{}\"\n",
"store.db.path = \"{}/sqlite.db\"\n"
),
temp_dir.path.display(),
temp_dir.path.display()
);
let config_file = CONFIG.replace("PATH", &temp_dir.path.to_string_lossy());
//let db: Store = SqliteStore::open(&Config::new(&config_file).unwrap())
let db: Store = FdbStore::open(&Config::new(&config_file).unwrap())
//let db: Store = FdbStore::open(&Config::new(&config_file).unwrap())
let db: Store = PostgresStore::open(&Config::new(&config_file).unwrap())
.await
.unwrap()
.into();