708 attach timestamp to deletion record and use it to find newest rec…

…ords (#746)

* [708] Pass timestamp to write and delete operations inside pearl holder

* [708] Update changelog

* [708] Fix changelog after merge

* [708] Update pearl version

CHANGELOG.md

@@ -8,6 +8,7 @@ Bob versions changelog
 
 #### Changed
 - Use cargo workspace to declare dependencies to avoid their duplication (#821)
+- Record timestamp is now passed to Pearl level and used to find newest record in get and exist functions (#708)
 
 #### Fixed
 

bob-backend/src/pearl/holder.rs

@@ -271,7 +271,7 @@ impl Holder {
         counter!(PEARL_PUT_COUNTER, 1);
         let data_size = Self::calc_data_size(&data);
         let timer = Instant::now();
-        let res = storage.write(key, data.to_serialized_bytes()).await;
+        let res = storage.write(key, data.to_serialized_bytes(), BlobRecordTimestamp::new(data.meta().timestamp())).await;
         let res = match res {
             Err(e) => {
                 counter!(PEARL_PUT_ERROR_COUNTER, 1);
@@ -505,15 +505,14 @@ impl Holder {
             .with_context(|| format!("cannot build pearl by path: {:?}", &self.inner.disk_path))
     }
 
-    pub async fn delete(&self, key: BobKey, _meta: &BobMeta, force_delete: bool) -> Result<u64, Error> {
+    pub async fn delete(&self, key: BobKey, meta: &BobMeta, force_delete: bool) -> Result<u64, Error> {
         let state = self.storage.read().await;
         if let Some(storage) = state.get() {
             trace!("Vdisk: {}, delete key: {}", self.inner.vdisk, key);
             counter!(PEARL_DELETE_COUNTER, 1);
             let timer = Instant::now();
-            // TODO: use meta
             let res = storage
-                .delete(Key::from(key), !force_delete)
+                .delete(Key::from(key), BlobRecordTimestamp::new(meta.timestamp()), !force_delete)
                 .await
                 .map_err(|e| {
                     trace!("error on delete: {:?}", e);