aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/login/mod.rs2
-rw-r--r--src/mail/incoming.rs74
-rw-r--r--src/storage/garage.rs6
-rw-r--r--src/storage/in_memory.rs6
-rw-r--r--src/storage/mod.rs3
5 files changed, 47 insertions, 44 deletions
diff --git a/src/login/mod.rs b/src/login/mod.rs
index f4bf4d2..a150829 100644
--- a/src/login/mod.rs
+++ b/src/login/mod.rs
@@ -47,7 +47,7 @@ pub struct Credentials {
#[derive(Clone, Debug)]
pub struct PublicCredentials {
/// The storage credentials are used to authenticate access to the underlying storage (S3, K2V)
- pub storage: StorageCredentials,
+ pub storage: Builders,
pub public_key: PublicKey,
}
diff --git a/src/mail/incoming.rs b/src/mail/incoming.rs
index 9899ae8..db22f3e 100644
--- a/src/mail/incoming.rs
+++ b/src/mail/incoming.rs
@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+//use std::collections::HashMap;
use std::convert::TryFrom;
use std::sync::{Arc, Weak};
@@ -6,11 +6,7 @@ use std::time::Duration;
use anyhow::{anyhow, bail, Result};
use futures::{future::BoxFuture, FutureExt};
-use k2v_client::{CausalityToken, K2vClient, K2vValue};
-use rusoto_s3::{
- DeleteObjectRequest, GetObjectRequest, ListObjectsV2Request, PutObjectRequest, S3Client, S3,
-};
-use tokio::io::AsyncReadExt;
+//use tokio::io::AsyncReadExt;
use tokio::sync::watch;
use tracing::{error, info, warn};
@@ -81,7 +77,7 @@ async fn incoming_mail_watch_process_internal(
tokio::select! {
inc_k = wait_new_mail => Some(inc_k),
- _ = tokio::time::sleep(MAIL_CHECK_INTERVAL) => Some(incoming_key),
+ _ = tokio::time::sleep(MAIL_CHECK_INTERVAL) => Some(k2v.from_orphan(incoming_key.to_orphan())),
_ = lock_held.changed() => None,
_ = rx_inbox_id.changed() => None,
}
@@ -220,7 +216,7 @@ fn k2v_lock_loop(k2v: storage::RowStore, pk: &'static str, sk: &'static str) ->
enum LockState {
Unknown,
Empty,
- Held(UniqueIdent, u64, CausalityToken),
+ Held(UniqueIdent, u64, storage::OrphanRowRef),
}
async fn k2v_lock_loop_internal(
@@ -236,10 +232,10 @@ async fn k2v_lock_loop_internal(
// Loop 1: watch state of lock in K2V, save that in corresponding watch channel
let watch_lock_loop: BoxFuture<Result<()>> = async {
- let mut ct = None;
+ let mut ct = k2v.row(pk, sk);
loop {
info!("k2v watch lock loop iter: ct = {:?}", ct);
- match k2v_wait_value_changed(&k2v, pk, sk, &ct).await {
+ match ct.poll().await {
Err(e) => {
error!(
"Error in k2v wait value changed: {} ; assuming we no longer hold lock.",
@@ -250,8 +246,8 @@ async fn k2v_lock_loop_internal(
}
Ok(cv) => {
let mut lock_state = None;
- for v in cv.value.iter() {
- if let K2vValue::Value(vbytes) = v {
+ for v in cv.content().iter() {
+ if let storage::Alternative::Value(vbytes) = v {
if vbytes.len() == 32 {
let ts = u64::from_be_bytes(vbytes[..8].try_into().unwrap());
let pid = UniqueIdent(vbytes[8..].try_into().unwrap());
@@ -264,16 +260,18 @@ async fn k2v_lock_loop_internal(
}
}
}
+ let new_ct = cv.to_ref();
+
info!(
"k2v watch lock loop: changed, old ct = {:?}, new ct = {:?}, v = {:?}",
- ct, cv.causality, lock_state
+ ct, new_ct, lock_state
);
state_tx.send(
lock_state
- .map(|(pid, ts)| LockState::Held(pid, ts, cv.causality.clone()))
+ .map(|(pid, ts)| LockState::Held(pid, ts, new_ct.to_orphan()))
.unwrap_or(LockState::Empty),
)?;
- ct = Some(cv.causality);
+ ct = new_ct;
}
}
}
@@ -359,7 +357,11 @@ async fn k2v_lock_loop_internal(
now_msec() + LOCK_DURATION.as_millis() as u64,
));
lock[8..].copy_from_slice(&our_pid.0);
- if let Err(e) = k2v.insert_item(pk, sk, lock, ct).await {
+ let row = match ct {
+ Some(orphan) => k2v.from_orphan(orphan),
+ None => k2v.row(pk, sk),
+ };
+ if let Err(e) = row.set_value(lock).push().await {
error!("Could not take lock: {}", e);
tokio::time::sleep(Duration::from_secs(30)).await;
}
@@ -385,7 +387,8 @@ async fn k2v_lock_loop_internal(
_ => None,
};
if let Some(ct) = release {
- let _ = k2v.delete_item(pk, sk, ct.clone()).await;
+ let row = k2v.from_orphan(ct);
+ let _ = row.rm().await;
}
}
@@ -407,13 +410,14 @@ impl EncryptedMessage {
}
pub async fn deliver_to(self: Arc<Self>, creds: PublicCredentials) -> Result<()> {
- let s3_client = creds.storage.s3_client()?;
- let k2v_client = creds.storage.k2v_client()?;
+ let s3_client = creds.storage.blob_store()?;
+ let k2v_client = creds.storage.row_store()?;
// Get causality token of previous watch key
- let watch_ct = match k2v_client.read_item(INCOMING_PK, INCOMING_WATCH_SK).await {
- Err(_) => None,
- Ok(cv) => Some(cv.causality),
+ let query = k2v_client.row(INCOMING_PK, INCOMING_WATCH_SK);
+ let watch_ct = match query.fetch().await {
+ Err(_) => query,
+ Ok(cv) => cv.to_ref(),
};
// Write mail to encrypted storage
@@ -421,28 +425,14 @@ impl EncryptedMessage {
sodiumoxide::crypto::sealedbox::seal(self.key.as_ref(), &creds.public_key);
let key_header = base64::encode(&encrypted_key);
- let por = PutObjectRequest {
- bucket: creds.storage.bucket.clone(),
- key: format!("incoming/{}", gen_ident()),
- metadata: Some(
- [(MESSAGE_KEY.to_string(), key_header)]
- .into_iter()
- .collect::<HashMap<_, _>>(),
- ),
- body: Some(self.encrypted_body.clone().into()),
- ..Default::default()
- };
- s3_client.put_object(por).await?;
+ let mut send = s3_client
+ .blob(&format!("incoming/{}", gen_ident()))
+ .set_value(self.encrypted_body.clone().into());
+ send.set_meta(MESSAGE_KEY, &key_header);
+ send.push().await?;
// Update watch key to signal new mail
- k2v_client
- .insert_item(
- INCOMING_PK,
- INCOMING_WATCH_SK,
- gen_ident().0.to_vec(),
- watch_ct,
- )
- .await?;
+ watch_ct.set_value(gen_ident().0.to_vec()).push().await?;
Ok(())
}
diff --git a/src/storage/garage.rs b/src/storage/garage.rs
index d6ac7ac..f458aeb 100644
--- a/src/storage/garage.rs
+++ b/src/storage/garage.rs
@@ -67,6 +67,12 @@ impl IRowRef for GrgRef {
}
}
+impl std::fmt::Debug for GrgRef {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ unimplemented!();
+ }
+}
+
impl IRowValue for GrgValue {
fn to_ref(&self) -> RowRef {
unimplemented!();
diff --git a/src/storage/in_memory.rs b/src/storage/in_memory.rs
index 0bdf9b1..20f96a4 100644
--- a/src/storage/in_memory.rs
+++ b/src/storage/in_memory.rs
@@ -72,6 +72,12 @@ impl IRowRef for MemRef {
}
}
+impl std::fmt::Debug for MemRef {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ unimplemented!();
+ }
+}
+
impl IRowValue for MemValue {
fn to_ref(&self) -> RowRef {
unimplemented!();
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
index c9a49c5..08ccfec 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -91,7 +91,7 @@ pub trait IRowStore
}
pub type RowStore = Box<dyn IRowStore + Sync + Send>;
-pub trait IRowRef
+pub trait IRowRef: std::fmt::Debug
{
/*fn clone_boxed(&self) -> RowRef;*/
fn to_orphan(&self) -> OrphanRowRef;
@@ -138,6 +138,7 @@ pub type BlobRef = Box<dyn IBlobRef + Send + Sync>;
pub trait IBlobValue {
fn to_ref(&self) -> BlobRef;
fn get_meta(&self, key: &str) -> Option<&[u8]>;
+ fn set_meta(&mut self, key: &str, val: &str);
fn content(&self) -> Option<&[u8]>;
fn push(&self) -> AsyncResult<()>;
}
generated by cgit v1.2.3 (git 2.25.1) at 2024-09-20 05:56:23 +0000