aboutsummaryrefslogtreecommitdiff
path: root/src/bayou.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/bayou.rs')
-rw-r--r--src/bayou.rs134
1 files changed, 63 insertions, 71 deletions
diff --git a/src/bayou.rs b/src/bayou.rs
index 3201783..f95bd82 100644
--- a/src/bayou.rs
+++ b/src/bayou.rs
@@ -9,16 +9,12 @@ use serde::{Deserialize, Serialize};
use tokio::io::AsyncReadExt;
use tokio::sync::{watch, Notify};
-use k2v_client::{BatchDeleteOp, BatchReadOp, CausalityToken, Filter, K2vClient, K2vValue};
-use rusoto_s3::{
- DeleteObjectRequest, GetObjectRequest, ListObjectsV2Request, PutObjectRequest, S3Client, S3,
-};
-
use crate::cryptoblob::*;
use crate::login::Credentials;
use crate::timestamp::*;
use crate::storage;
+
const KEEP_STATE_EVERY: usize = 64;
// Checkpointing interval constants: a checkpoint is not made earlier
@@ -61,7 +57,7 @@ pub struct Bayou<S: BayouState> {
last_try_checkpoint: Option<Instant>,
watch: Arc<K2vWatch>,
- last_sync_watch_ct: Option<CausalityToken>,
+ last_sync_watch_ct: storage::RowRef,
}
impl<S: BayouState> Bayou<S> {
@@ -69,6 +65,7 @@ impl<S: BayouState> Bayou<S> {
let k2v_client = creds.row_client()?;
let s3_client = creds.blob_client()?;
+ let target = k2v_client.row(&path, WATCH_SK);
let watch = K2vWatch::new(creds, path.clone(), WATCH_SK.to_string())?;
Ok(Self {
@@ -81,7 +78,7 @@ impl<S: BayouState> Bayou<S> {
last_sync: None,
last_try_checkpoint: None,
watch,
- last_sync_watch_ct: None,
+ last_sync_watch_ct: target,
})
}
@@ -134,7 +131,7 @@ impl<S: BayouState> Bayou<S> {
// 3. List all operations starting from checkpoint
let ts_ser = self.checkpoint.0.to_string();
debug!("(sync) looking up operations starting at {}", ts_ser);
- let ops_map = self.k2v.select(storage::Selector::Range { begin: &ts_ser, end: WATCH_SK }).await?;
+ let ops_map = self.k2v.select(storage::Selector::Range { shard_key: &self.path, begin: &ts_ser, end: WATCH_SK }).await?;
/*let ops_map = self
.k2v
.read_batch(&[BatchReadOp {
@@ -158,18 +155,22 @@ impl<S: BayouState> Bayou<S> {
let mut ops = vec![];
for row_value in ops_map {
- let ts = row_value.timestamp();
- if val.value.len() != 1 {
- bail!("Invalid operation, has {} values", val.value.len());
+ let row = row_value.to_ref();
+ let sort_key = row.key().1;
+ let ts = sort_key.parse::<Timestamp>().map_err(|_| anyhow!("Invalid operation timestamp: {}", sort_key))?;
+
+ let val = row_value.content();
+ if val.len() != 1 {
+ bail!("Invalid operation, has {} values", row_value.content().len());
}
- match &val.value[0] {
- K2vValue::Value(v) => {
+ match &val[0] {
+ storage::Alternative::Value(v) => {
let op = open_deserialize::<S::Op>(v, &self.key)?;
- debug!("(sync) operation {}: {} {:?}", tsstr, base64::encode(v), op);
+ debug!("(sync) operation {}: {} {:?}", sort_key, base64::encode(v), op);
ops.push((ts, op));
}
- K2vValue::Tombstone => {
- unreachable!();
+ storage::Alternative::Tombstone => {
+ continue;
}
}
}
@@ -372,13 +373,12 @@ impl<S: BayouState> Bayou<S> {
let cryptoblob = seal_serialize(&state_cp, &self.key)?;
debug!("(cp) checkpoint body length: {}", cryptoblob.len());
- let por = PutObjectRequest {
- bucket: self.bucket.clone(),
- key: format!("{}/checkpoint/{}", self.path, ts_cp.to_string()),
- body: Some(cryptoblob.into()),
- ..Default::default()
- };
- self.s3.put_object(por).await?;
+ self.s3
+ .blob(format!("{}/checkpoint/{}", self.path, ts_cp.to_string()).as_str())
+ .set_value(cryptoblob.into())
+ .push()
+ .await?;
+
// Drop old checkpoints (but keep at least CHECKPOINTS_TO_KEEP of them)
let ecp_len = existing_checkpoints.len();
@@ -388,25 +388,22 @@ impl<S: BayouState> Bayou<S> {
// Delete blobs
for (_ts, key) in existing_checkpoints[..last_to_keep].iter() {
debug!("(cp) drop old checkpoint {}", key);
- let dor = DeleteObjectRequest {
- bucket: self.bucket.clone(),
- key: key.to_string(),
- ..Default::default()
- };
- self.s3.delete_object(dor).await?;
+ self.s3
+ .blob(key)
+ .rm()
+ .await?;
}
// Delete corresponding range of operations
let ts_ser = existing_checkpoints[last_to_keep].0.to_string();
self.k2v
- .delete_batch(&[BatchDeleteOp {
- partition_key: &self.path,
- prefix: None,
- start: None,
- end: Some(&ts_ser),
- single_item: false,
- }])
+ .rm(storage::Selector::Range{
+ shard_key: &self.path,
+ begin: "",
+ end: &ts_ser
+ })
.await?;
+
}
Ok(())
@@ -425,22 +422,14 @@ impl<S: BayouState> Bayou<S> {
async fn list_checkpoints(&self) -> Result<Vec<(Timestamp, String)>> {
let prefix = format!("{}/checkpoint/", self.path);
- let lor = ListObjectsV2Request {
- bucket: self.bucket.clone(),
- max_keys: Some(1000),
- prefix: Some(prefix.clone()),
- ..Default::default()
- };
-
- let checkpoints_res = self.s3.list_objects_v2(lor).await?;
+ let checkpoints_res = self.s3.list(&prefix).await?;
let mut checkpoints = vec![];
- for object in checkpoints_res.contents.unwrap_or_default() {
- if let Some(key) = object.key {
- if let Some(ckid) = key.strip_prefix(&prefix) {
- if let Ok(ts) = ckid.parse::<Timestamp>() {
- checkpoints.push((ts, key));
- }
+ for object in checkpoints_res {
+ let key = object.key();
+ if let Some(ckid) = key.strip_prefix(&prefix) {
+ if let Ok(ts) = ckid.parse::<Timestamp>() {
+ checkpoints.push((ts, key.into()));
}
}
}
@@ -454,23 +443,25 @@ impl<S: BayouState> Bayou<S> {
struct K2vWatch {
pk: String,
sk: String,
- rx: watch::Receiver<Option<CausalityToken>>,
+ rx: watch::Receiver<storage::RowRef>,
notify: Notify,
}
impl K2vWatch {
/// Creates a new watch and launches subordinate threads.
/// These threads hold Weak pointers to the struct;
- /// the exit when the Arc is dropped.
+ /// they exit when the Arc is dropped.
fn new(creds: &Credentials, pk: String, sk: String) -> Result<Arc<Self>> {
- let (tx, rx) = watch::channel::<Option<CausalityToken>>(None);
+ let row_client = creds.row_client()?;
+
+ let (tx, rx) = watch::channel::<storage::RowRef>(row_client.row(&pk, &sk));
let notify = Notify::new();
let watch = Arc::new(K2vWatch { pk, sk, rx, notify });
tokio::spawn(Self::background_task(
Arc::downgrade(&watch),
- creds.k2v_client()?,
+ row_client,
tx,
));
@@ -479,41 +470,42 @@ impl K2vWatch {
async fn background_task(
self_weak: Weak<Self>,
- k2v: K2vClient,
- tx: watch::Sender<Option<CausalityToken>>,
+ k2v: storage::RowStore,
+ tx: watch::Sender<storage::RowRef>,
) {
- let mut ct = None;
+ let mut row = match Weak::upgrade(&self_weak) {
+ Some(this) => k2v.row(&this.pk, &this.sk),
+ None => {
+ error!("can't start loop");
+ return
+ },
+ };
+
while let Some(this) = Weak::upgrade(&self_weak) {
debug!(
- "bayou k2v watch bg loop iter ({}, {}): ct = {:?}",
- this.pk, this.sk, ct
+ "bayou k2v watch bg loop iter ({}, {})",
+ this.pk, this.sk
);
tokio::select!(
_ = tokio::time::sleep(Duration::from_secs(60)) => continue,
- update = k2v_wait_value_changed(&k2v, &this.pk, &this.sk, &ct) => {
+ update = row.poll() => {
+ //update = k2v_wait_value_changed(&k2v, &this.pk, &this.sk, &ct) => {
match update {
Err(e) => {
error!("Error in bayou k2v wait value changed: {}", e);
tokio::time::sleep(Duration::from_secs(30)).await;
}
- Ok(cv) => {
- if tx.send(Some(cv.causality.clone())).is_err() {
+ Ok(new_value) => {
+ row = new_value.to_ref();
+ if tx.send(XXX).is_err() {
break;
}
- ct = Some(cv.causality);
}
}
}
_ = this.notify.notified() => {
let rand = u128::to_be_bytes(thread_rng().gen()).to_vec();
- if let Err(e) = k2v
- .insert_item(
- &this.pk,
- &this.sk,
- rand,
- ct.clone(),
- )
- .await
+ if let Err(e) = row.set_value(rand).push().await
{
error!("Error in bayou k2v watch updater loop: {}", e);
tokio::time::sleep(Duration::from_secs(30)).await;